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 0f2d66b5ca0fa88641b8047c91cf9fd3459e16e3 Author: Tian Jiang <[email protected]> AuthorDate: Thu Aug 13 16:36:54 2026 +0800 ver1 --- .../apache/iotdb/db/i18n/ImportWALMessages.java | 80 ++ .../apache/iotdb/db/i18n/ImportWALMessages.java | 79 ++ .../storageengine/dataregion/wal/io/WALReader.java | 9 + .../java/org/apache/iotdb/db/tools/ImportWAL.java | 977 +++++++++++++++++++++ .../org/apache/iotdb/db/tools/ImportWALTest.java | 506 +++++++++++ scripts/tools/import-wal.sh | 52 ++ scripts/tools/windows/import-wal.bat | 43 + 7 files changed, 1746 insertions(+) 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 new file mode 100644 index 00000000000..2ed53656ec0 --- /dev/null +++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/ImportWALMessages.java @@ -0,0 +1,80 @@ +/* + * 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.db.i18n; + +/** Compile-time i18n constants for the WAL import tool (English). */ +public final class ImportWALMessages { + + public static final String MESSAGE_IMPORT_WAL_5E42804E = "import-wal"; + public static final String + MESSAGE_PATH_OF_A_WAL_FILE_OR_A_DIRECTORY_CONTAINING_WAL_FILES_473D0554 = + "Path of a WAL file or a directory containing WAL files."; + public static final String MESSAGE_TARGET_IOTDB_HOST_DEFAULT_127_0_0_1_3729156F = + "Target IoTDB host. Default: 127.0.0.1."; + public static final String MESSAGE_TARGET_IOTDB_RPC_PORT_DEFAULT_6667_FC0D345D = + "Target IoTDB RPC port. Default: 6667."; + public static final String MESSAGE_TARGET_IOTDB_USERNAME_DEFAULT_ROOT_EB91453B = + "Target IoTDB username. Default: root."; + public static final String + MESSAGE_TARGET_IOTDB_PASSWORD_PROMPTED_INTERACTIVELY_IF_OMITTED_29681961 = + "Target IoTDB password. Prompted interactively if omitted."; + 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_PRINT_THIS_HELP_MESSAGE_E800AF7A = + "Print this help message."; + public static final String MESSAGE_ARGUMENT_ERROR_ARG_A9767F62 = "Argument error: %s"; + public static final String MESSAGE_WAL_IMPORT_FAILED_ARG_55C014BA = "WAL import failed: %s"; + public static final String EXCEPTION_SOURCE_PATH_DOES_NOT_EXIST_ARG_7C806CA2 = + "Source path does not exist: %s"; + public static final String EXCEPTION_SOURCE_FILE_IS_NOT_A_WAL_FILE_ARG_14A43F76 = + "Source file is not a WAL file: %s"; + public static final String EXCEPTION_NO_WAL_FILES_FOUND_UNDER_ARG_45F7FA22 = + "No WAL files found under: %s"; + public static final String EXCEPTION_INVALID_PORT_ARG_A7CDD5AC = "Invalid port: %s"; + public static final String + MESSAGE_REPLAYED_ARG_OPERATIONS_FROM_ARG_WAL_FILES_SKIPPED_ARG_ENTRIES_F0D37E3A = + "Replayed %d operations from %d WAL files; skipped %d entries."; + public static final String + MESSAGE_PROGRESS_ARG_COMPLETED_FILES_ARG_TOTAL_FILES_ARG_PROCESSED_BYTES_ARG_TOTAL_BYTES_ARG_PERCENT_ARG_ELAPSED_SECONDS_ARG_RATE_ARG_MB_PER_SECOND_F1C1356F = + "Progress: %d/%d WAL files completed, %d/%d bytes (%.1f%%), elapsed %.1f s, rate %.1f MB/s."; + 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 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_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 = + "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 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 = + "Unsupported snapshot data type: %s"; + public static final String EXCEPTION_THE_WAL_FILE_IS_TRUNCATED_OR_CORRUPTED_6B0734C5 = + "The WAL file is truncated or corrupted."; + public static final String + EXCEPTION_PASSWORD_WAS_NOT_PROVIDED_AND_INTERACTIVE_INPUT_IS_UNAVAILABLE_40F42BCD = + "Password was not provided and interactive input is unavailable. Specify -pw/--password."; + + private ImportWALMessages() {} +} 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 new file mode 100644 index 00000000000..c8dc2266dc8 --- /dev/null +++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/ImportWALMessages.java @@ -0,0 +1,79 @@ +/* + * 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.db.i18n; + +/** WAL 导入工具的编译期国际化常量（中文）。 */ +public final class ImportWALMessages { + + public static final String MESSAGE_IMPORT_WAL_5E42804E = "import-wal"; + public static final String + MESSAGE_PATH_OF_A_WAL_FILE_OR_A_DIRECTORY_CONTAINING_WAL_FILES_473D0554 = + "WAL 文件或包含 WAL 文件的目录路径。"; + public static final String MESSAGE_TARGET_IOTDB_HOST_DEFAULT_127_0_0_1_3729156F = + "目标 IoTDB 主机。默认：127.0.0.1。"; + public static final String MESSAGE_TARGET_IOTDB_RPC_PORT_DEFAULT_6667_FC0D345D = + "目标 IoTDB RPC 端口。默认：6667。"; + public static final String MESSAGE_TARGET_IOTDB_USERNAME_DEFAULT_ROOT_EB91453B = + "目标 IoTDB 用户名。默认：root。"; + public static final String + MESSAGE_TARGET_IOTDB_PASSWORD_PROMPTED_INTERACTIVELY_IF_OMITTED_29681961 = + "目标 IoTDB 密码。未提供时将交互式询问。"; + 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_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"; + public static final String EXCEPTION_SOURCE_PATH_DOES_NOT_EXIST_ARG_7C806CA2 = + "源路径不存在：%s"; + public static final String EXCEPTION_SOURCE_FILE_IS_NOT_A_WAL_FILE_ARG_14A43F76 = + "源文件不是 WAL 文件：%s"; + public static final String EXCEPTION_NO_WAL_FILES_FOUND_UNDER_ARG_45F7FA22 = + "路径下未找到 WAL 文件：%s"; + public static final String EXCEPTION_INVALID_PORT_ARG_A7CDD5AC = "无效端口：%s"; + public static final String + MESSAGE_REPLAYED_ARG_OPERATIONS_FROM_ARG_WAL_FILES_SKIPPED_ARG_ENTRIES_F0D37E3A = + "已重放 %d 个操作（来自 %d 个 WAL 文件）；跳过 %d 个条目。"; + public static final String + MESSAGE_PROGRESS_ARG_COMPLETED_FILES_ARG_TOTAL_FILES_ARG_PROCESSED_BYTES_ARG_TOTAL_BYTES_ARG_PERCENT_ARG_ELAPSED_SECONDS_ARG_RATE_ARG_MB_PER_SECOND_F1C1356F = + "进度：已完成 %d/%d 个 WAL 文件，已处理 %d/%d 字节（%.1f%%），耗时 %.1f 秒，速率 %.1f MB/s。"; + 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 EXCEPTION_FAILED_TO_REPLAY_WAL_FILE_ARG_AT_OFFSET_ARG_ARG_FCFAF7F9 = + "重放 WAL 文件 %s 时失败，偏移量 %d：%s"; + 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 = + "不支持的 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 EXCEPTION_INSERT_NODE_ARG_CONTAINS_NO_REPLAYABLE_DATA_5DA13453 = + "Insert node %s 不包含可重放数据。"; + public static final String EXCEPTION_UNSUPPORTED_SNAPSHOT_DATA_TYPE_ARG_7A32D312 = + "Unsupported snapshot data type: %s"; + public static final String EXCEPTION_THE_WAL_FILE_IS_TRUNCATED_OR_CORRUPTED_6B0734C5 = + "WAL 文件被截断或已损坏。"; + public static final String + EXCEPTION_PASSWORD_WAS_NOT_PROVIDED_AND_INTERACTIVE_INPUT_IS_UNAVAILABLE_40F42BCD = + "未提供密码且当前环境不支持交互式输入，请指定 -pw/--password。"; + + private ImportWALMessages() {} +} diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALReader.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALReader.java index befcf58c632..0a2c994eb54 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALReader.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALReader.java @@ -72,6 +72,11 @@ public class WALReader implements Closeable { return false; } try { + // An active WAL has no end marker until its writer closes. Reaching EOF exactly between + // entries is therefore valid, while EOF during deserialization still marks a partial entry. + if (walInputStream.available() == 0) { + return false; + } nextEntry = WALEntry.deserialize(logStream); if (nextEntry.getType() == WALEntryType.WAL_FILE_INFO_END_MARKER) { nextEntry = null; @@ -98,6 +103,10 @@ public class WALReader implements Closeable { return walInputStream.getFileCurrentPos(); } + public boolean isFileCorrupted() { + return fileCorrupted; + } + /** * Like {@link Iterator#next()}. * 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 new file mode 100644 index 00000000000..f8316499bdb --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/tools/ImportWAL.java @@ -0,0 +1,977 @@ +/* + * 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.db.tools; + +import org.apache.iotdb.commons.path.MeasurementPath; +import org.apache.iotdb.db.i18n.ImportWALMessages; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.DeleteDataNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode; +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.RelationalInsertRowNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowsNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertTabletNode; +import org.apache.iotdb.db.storageengine.dataregion.memtable.AlignedWritableMemChunk; +import org.apache.iotdb.db.storageengine.dataregion.memtable.IMemTable; +import org.apache.iotdb.db.storageengine.dataregion.memtable.IWritableMemChunk; +import org.apache.iotdb.db.storageengine.dataregion.memtable.IWritableMemChunkGroup; +import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntry; +import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntryType; +import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALReader; +import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALFileUtils; +import org.apache.iotdb.db.subscription.broker.consensus.ConsensusLogToTabletConverter; +import org.apache.iotdb.db.subscription.columnfilter.ColumnFilterMatcher; +import org.apache.iotdb.db.utils.datastructure.AlignedTVList; +import org.apache.iotdb.db.utils.datastructure.TVList; +import org.apache.iotdb.isession.SessionDataSet; +import org.apache.iotdb.rpc.IoTDBConnectionException; +import org.apache.iotdb.rpc.StatementExecutionException; +import org.apache.iotdb.session.Session; + +import org.apache.commons.cli.CommandLine; +import org.apache.commons.cli.DefaultParser; +import org.apache.commons.cli.HelpFormatter; +import org.apache.commons.cli.Option; +import org.apache.commons.cli.Options; +import org.apache.commons.cli.ParseException; +import org.apache.tsfile.common.conf.TSFileConfig; +import org.apache.tsfile.enums.ColumnCategory; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.metadata.IDeviceID; +import org.apache.tsfile.utils.Binary; +import org.apache.tsfile.utils.BitMap; +import org.apache.tsfile.utils.DateUtils; +import org.apache.tsfile.write.record.Tablet; +import org.apache.tsfile.write.schema.IMeasurementSchema; +import org.apache.tsfile.write.schema.MeasurementSchema; + +import java.io.Console; +import java.io.IOException; +import java.io.PrintStream; +import java.io.PrintWriter; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.Paths; +import java.time.LocalDate; +import java.util.ArrayList; +import java.util.Collections; +import java.util.Comparator; +import java.util.HashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Objects; +import java.util.stream.Collectors; +import java.util.stream.Stream; + +public class ImportWAL { + + private static final int CODE_OK = 0; + private static final int CODE_ERROR = 1; + private static final String DEFAULT_HOST = "127.0.0.1"; + private static final int DEFAULT_PORT = 6667; + private static final String DEFAULT_USER = "root"; + private static final int SNAPSHOT_TABLET_ROW_LIMIT = 1024; + + private ImportWAL() {} + + public static void main(final String[] args) { + System.exit(run(args, System.out, System.err)); + } + + static int run(final String[] args, final PrintStream out, final PrintStream err) { + final Options options = createOptions(); + if (containsHelpOption(args)) { + printHelp(options, out); + return CODE_OK; + } + + final CommandLine commandLine; + try { + commandLine = new DefaultParser().parse(options, args); + } catch (final ParseException e) { + err.printf(ImportWALMessages.MESSAGE_ARGUMENT_ERROR_ARG_A9767F62, e.getMessage()); + err.println(); + printHelp(options, err); + return CODE_ERROR; + } + + try { + final Path source = Paths.get(commandLine.getOptionValue("file")); + final List<Path> walFiles = collectWALFiles(source); + final String database = commandLine.getOptionValue("database"); + final String password = getPassword(commandLine); + final Session treeSession = + createSession( + commandLine.getOptionValue("host", DEFAULT_HOST), + parsePort(commandLine.getOptionValue("port", String.valueOf(DEFAULT_PORT))), + commandLine.getOptionValue("username", DEFAULT_USER), + password, + null); + final Session tableSession = + database == null + ? null + : createSession( + commandLine.getOptionValue("host", DEFAULT_HOST), + parsePort(commandLine.getOptionValue("port", String.valueOf(DEFAULT_PORT))), + commandLine.getOptionValue("username", DEFAULT_USER), + password, + database); + try { + treeSession.open(false); + if (tableSession != null) { + tableSession.open(false); + } + final ReplayStatistics statistics = + replayWALFiles(walFiles, new WALReplayer(treeSession, tableSession, database), out); + out.printf( + ImportWALMessages + .MESSAGE_REPLAYED_ARG_OPERATIONS_FROM_ARG_WAL_FILES_SKIPPED_ARG_ENTRIES_F0D37E3A, + statistics.replayedOperationCount, + walFiles.size(), + statistics.skippedEntryCount); + out.println(); + out.printf( + ImportWALMessages + .MESSAGE_IMPORT_DURATION_ARG_SECONDS_TOTAL_SIZE_ARG_BYTES_AVERAGE_RATE_ARG_MB_PER_SECOND_4B4EA58D, + statistics.getElapsedSeconds(), + statistics.getTotalBytes(), + statistics.getAverageRateMbPerSecond()); + out.println(); + } finally { + closeSession(tableSession); + closeSession(treeSession); + } + return CODE_OK; + } catch (final Exception e) { + err.printf(ImportWALMessages.MESSAGE_WAL_IMPORT_FAILED_ARG_55C014BA, e.getMessage()); + err.println(); + return CODE_ERROR; + } + } + + private static Options createOptions() { + final Options options = new Options(); + options.addOption( + Option.builder("f") + .longOpt("file") + .hasArg() + .required() + .desc( + ImportWALMessages + .MESSAGE_PATH_OF_A_WAL_FILE_OR_A_DIRECTORY_CONTAINING_WAL_FILES_473D0554) + .build()); + options.addOption( + Option.builder("h") + .longOpt("host") + .hasArg() + .desc(ImportWALMessages.MESSAGE_TARGET_IOTDB_HOST_DEFAULT_127_0_0_1_3729156F) + .build()); + options.addOption( + Option.builder("p") + .longOpt("port") + .hasArg() + .desc(ImportWALMessages.MESSAGE_TARGET_IOTDB_RPC_PORT_DEFAULT_6667_FC0D345D) + .build()); + options.addOption( + Option.builder("u") + .longOpt("username") + .hasArg() + .desc(ImportWALMessages.MESSAGE_TARGET_IOTDB_USERNAME_DEFAULT_ROOT_EB91453B) + .build()); + options.addOption( + Option.builder("pw") + .longOpt("password") + .hasArg() + .desc( + ImportWALMessages + .MESSAGE_TARGET_IOTDB_PASSWORD_PROMPTED_INTERACTIVELY_IF_OMITTED_29681961) + .build()); + options.addOption( + Option.builder("db") + .longOpt("database") + .hasArg() + .desc(ImportWALMessages.MESSAGE_TARGET_DATABASE_FOR_TABLE_MODEL_WAL_ENTRIES_27BACD1C) + .build()); + options.addOption( + Option.builder() + .longOpt("help") + .desc(ImportWALMessages.MESSAGE_PRINT_THIS_HELP_MESSAGE_E800AF7A) + .build()); + return options; + } + + private static String getPassword(final CommandLine commandLine) { + return getPassword(commandLine, System.console()); + } + + static String getPassword(final CommandLine commandLine, final Console console) { + if (commandLine.hasOption("password")) { + return commandLine.getOptionValue("password"); + } + if (console == null) { + throw new IllegalArgumentException( + ImportWALMessages + .EXCEPTION_PASSWORD_WAS_NOT_PROVIDED_AND_INTERACTIVE_INPUT_IS_UNAVAILABLE_40F42BCD); + } + final char[] password = + console.readPassword(ImportWALMessages.MESSAGE_PASSWORD_PROMPT_F2D0E794); + if (password == null) { + throw new IllegalArgumentException( + ImportWALMessages + .EXCEPTION_PASSWORD_WAS_NOT_PROVIDED_AND_INTERACTIVE_INPUT_IS_UNAVAILABLE_40F42BCD); + } + return new String(password); + } + + private static boolean containsHelpOption(final String[] args) { + if (args == null) { + return false; + } + for (final String arg : args) { + if ("--help".equals(arg) || "-help".equals(arg)) { + return true; + } + } + return false; + } + + private static void printHelp(final Options options, final PrintStream stream) { + final HelpFormatter formatter = new HelpFormatter(); + formatter.setWidth(120); + formatter.printHelp( + new PrintWriter(stream, true), + 120, + ImportWALMessages.MESSAGE_IMPORT_WAL_5E42804E, + null, + options, + 2, + 2, + null, + true); + } + + private static int parsePort(final String port) { + try { + final int value = Integer.parseInt(port); + if (value <= 0 || value > 65535) { + throw new NumberFormatException(port); + } + return value; + } catch (final NumberFormatException e) { + throw new IllegalArgumentException( + String.format(ImportWALMessages.EXCEPTION_INVALID_PORT_ARG_A7CDD5AC, port), e); + } + } + + private static Session createSession( + final String host, + final int port, + final String username, + final String password, + final String database) { + final Session.Builder builder = + new Session.Builder().host(host).port(port).username(username).password(password); + if (database != null) { + builder.sqlDialect("table").database(database); + } + return builder.build(); + } + + private static void closeSession(final Session session) { + if (session == null) { + return; + } + try { + session.close(); + } catch (final IoTDBConnectionException ignored) { + // The import result has already been determined; closing failure must not hide it. + } + } + + static List<Path> collectWALFiles(final Path source) throws IOException { + if (!Files.exists(source)) { + throw new IOException( + String.format( + ImportWALMessages.EXCEPTION_SOURCE_PATH_DOES_NOT_EXIST_ARG_7C806CA2, source)); + } + if (Files.isRegularFile(source)) { + if (!isWALFile(source)) { + throw new IOException( + String.format( + ImportWALMessages.EXCEPTION_SOURCE_FILE_IS_NOT_A_WAL_FILE_ARG_14A43F76, source)); + } + return List.of(source.toAbsolutePath().normalize()); + } + if (!Files.isDirectory(source)) { + throw new IOException( + String.format( + ImportWALMessages.EXCEPTION_SOURCE_PATH_DOES_NOT_EXIST_ARG_7C806CA2, source)); + } + + final List<Path> walFiles; + try (Stream<Path> stream = Files.walk(source)) { + walFiles = + stream + .filter(Files::isRegularFile) + .filter(ImportWAL::isWALFile) + .map(path -> path.toAbsolutePath().normalize()) + .sorted(WAL_FILE_COMPARATOR) + .collect(Collectors.toList()); + } + if (walFiles.isEmpty()) { + throw new IOException( + String.format(ImportWALMessages.EXCEPTION_NO_WAL_FILES_FOUND_UNDER_ARG_45F7FA22, source)); + } + return walFiles; + } + + private static boolean isWALFile(final Path path) { + return path.getFileName().toString().toLowerCase(Locale.ROOT).endsWith(".wal"); + } + + private static final Comparator<Path> WAL_FILE_COMPARATOR = + Comparator.comparing((Path path) -> Objects.toString(path.getParent(), "")) + .thenComparingLong(ImportWAL::getWALVersion) + .thenComparing(path -> path.getFileName().toString()); + + private static long getWALVersion(final Path path) { + final String filename = path.getFileName().toString(); + return WALFileUtils.WAL_FILE_NAME_PATTERN.matcher(filename).find() + ? WALFileUtils.parseVersionId(filename) + : Long.MAX_VALUE; + } + + static ReplayStatistics replayWALFiles(final List<Path> walFiles, final WALReplayer replayer) + throws IOException { + return replayWALFiles(walFiles, replayer, null); + } + + static ReplayStatistics replayWALFiles( + final List<Path> walFiles, final WALReplayer replayer, final PrintStream progressStream) + throws IOException { + final ReplayStatistics statistics = new ReplayStatistics(); + final long startNanos = System.nanoTime(); + for (final Path walFile : walFiles) { + statistics.totalBytes += Files.size(walFile); + } + for (final Path walFile : walFiles) { + try (WALReader reader = new WALReader(walFile.toFile())) { + long offset = reader.getWALCurrentReadOffset(); + while (reader.hasNext()) { + final WALEntry entry = reader.next(); + try { + if (replayer.replay(entry)) { + statistics.replayedOperationCount++; + } else { + statistics.skippedEntryCount++; + } + } catch (final IoTDBConnectionException | StatementExecutionException e) { + throw new WALReplayException( + String.format( + ImportWALMessages + .EXCEPTION_FAILED_TO_REPLAY_WAL_FILE_ARG_AT_OFFSET_ARG_ARG_FCFAF7F9, + walFile, + offset, + e.getMessage()), + e); + } + offset = reader.getWALCurrentReadOffset(); + } + if (reader.isFileCorrupted()) { + throw new WALReplayException( + String.format( + ImportWALMessages + .EXCEPTION_FAILED_TO_REPLAY_WAL_FILE_ARG_AT_OFFSET_ARG_ARG_FCFAF7F9, + walFile, + reader.getWALCurrentReadOffset(), + ImportWALMessages.EXCEPTION_THE_WAL_FILE_IS_TRUNCATED_OR_CORRUPTED_6B0734C5), + null); + } + statistics.completedFileCount++; + statistics.processedBytes += Files.size(walFile); + statistics.elapsedNanos = System.nanoTime() - startNanos; + if (progressStream != null) { + progressStream.printf( + ImportWALMessages + .MESSAGE_PROGRESS_ARG_COMPLETED_FILES_ARG_TOTAL_FILES_ARG_PROCESSED_BYTES_ARG_TOTAL_BYTES_ARG_PERCENT_ARG_ELAPSED_SECONDS_ARG_RATE_ARG_MB_PER_SECOND_F1C1356F, + statistics.completedFileCount, + walFiles.size(), + statistics.processedBytes, + statistics.totalBytes, + statistics.getProgressPercent(), + statistics.getElapsedSeconds(), + statistics.getAverageRateMbPerSecond()); + progressStream.println(); + } + } catch (final IOException e) { + if (e instanceof WALReplayException walReplayException) { + throw walReplayException; + } + throw new WALReplayException( + String.format( + ImportWALMessages + .EXCEPTION_FAILED_TO_REPLAY_WAL_FILE_ARG_AT_OFFSET_ARG_ARG_FCFAF7F9, + walFile, + 0, + e.getMessage()), + e); + } + } + return statistics; + } + + static class WALReplayer { + + private final Session treeSession; + private final Session tableSession; + private final ConsensusLogToTabletConverter converter; + private final UnsupportedEntryPrompt unsupportedEntryPrompt; + private final Map<String, List<IMeasurementSchema>> tableTagSchemas = new HashMap<>(); + + WALReplayer( + final Session treeSession, final Session tableSession, final String tableDatabaseName) { + this( + treeSession, + tableSession, + tableDatabaseName, + createUnsupportedEntryPrompt(System.console())); + } + + WALReplayer( + final Session treeSession, + final Session tableSession, + final String tableDatabaseName, + final UnsupportedEntryPrompt unsupportedEntryPrompt) { + this.treeSession = treeSession; + this.tableSession = tableSession; + this.unsupportedEntryPrompt = unsupportedEntryPrompt; + converter = + new ConsensusLogToTabletConverter( + null, null, ColumnFilterMatcher.matchAll(), tableDatabaseName); + } + + boolean replay(final WALEntry entry) + throws IoTDBConnectionException, StatementExecutionException { + if (entry.getType() == WALEntryType.MEMORY_TABLE_SNAPSHOT + || entry.getType() == WALEntryType.OLD_MEMORY_TABLE_SNAPSHOT) { + return replayMemTableSnapshot((IMemTable) entry.getValue()); + } + if (entry.getValue() instanceof InsertNode insertNode) { + replayInsert(insertNode); + return true; + } + if (entry.getValue() instanceof DeleteDataNode deleteDataNode) { + 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)) { + return false; + } + throw unsupportedOperation(entry); + } + return false; + } + + private static UnsupportedEntryPrompt createUnsupportedEntryPrompt(final Console console) { + if (console == null) { + return null; + } + return entry -> { + 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())); + } + + @FunctionalInterface + interface UnsupportedEntryPrompt { + + boolean shouldSkip(WALEntry entry); + } + + private static StatementExecutionException unsupportedOperation(final WALEntry entry) { + return new StatementExecutionException( + String.format( + ImportWALMessages.EXCEPTION_UNSUPPORTED_WAL_OPERATION_ARG_ABD227A0, entry.getType())); + } + + private void replayInsert(final InsertNode node) + throws IoTDBConnectionException, StatementExecutionException { + if (node instanceof InsertRowsNode insertRowsNode + && !(node instanceof RelationalInsertRowsNode)) { + for (final InsertNode rowNode : insertRowsNode.getInsertRowNodeList()) { + replayInsert(rowNode); + } + return; + } + final List<Tablet> tablets = converter.convert(node); + if (tablets.isEmpty()) { + throw new StatementExecutionException( + String.format( + ImportWALMessages.EXCEPTION_INSERT_NODE_ARG_CONTAINS_NO_REPLAYABLE_DATA_5DA13453, + node.getType())); + } + final boolean tableModel = isTableModelInsert(node); + if (tableModel && tableSession == null) { + throw new StatementExecutionException( + ImportWALMessages.EXCEPTION_TABLE_MODEL_WAL_ENTRIES_REQUIRE_DB_DATABASE_F7597726); + } + for (final Tablet tablet : tablets) { + if (tableModel) { + tableSession.insertRelationalTablet(tablet); + } else if (node.isAligned()) { + treeSession.insertAlignedTablet(tablet); + } else { + treeSession.insertTablet(tablet); + } + } + } + + private static boolean isTableModelInsert(final InsertNode node) { + return node instanceof RelationalInsertRowNode + || node instanceof RelationalInsertRowsNode + || node instanceof RelationalInsertTabletNode; + } + + private void replayTreeDelete(final DeleteDataNode node) + throws IoTDBConnectionException, StatementExecutionException { + final List<String> paths = new ArrayList<>(node.getPathList().size()); + for (final MeasurementPath path : node.getPathList()) { + paths.add(path.getFullPath()); + } + treeSession.deleteData(paths, node.getDeleteStartTime(), node.getDeleteEndTime()); + } + + private boolean replayMemTableSnapshot(final IMemTable memTable) + throws IoTDBConnectionException, StatementExecutionException { + if (memTable == null || memTable.isSignalMemTable()) { + return false; + } + boolean replayed = false; + for (Map.Entry<IDeviceID, IWritableMemChunkGroup> deviceEntry : + memTable.getMemTableMap().entrySet()) { + final IDeviceID deviceId = deviceEntry.getKey(); + final IWritableMemChunkGroup group = deviceEntry.getValue(); + for (IWritableMemChunk chunk : group.getMemChunkMap().values()) { + if (chunk == null || chunk.isEmpty()) { + continue; + } + if (chunk instanceof AlignedWritableMemChunk) { + replayed |= replayAlignedMemChunk(deviceId, (AlignedWritableMemChunk) chunk); + } else { + replayed |= replayNonAlignedMemChunk(deviceId, chunk); + } + } + } + return replayed; + } + + private boolean replayNonAlignedMemChunk( + final IDeviceID deviceId, final IWritableMemChunk chunk) + throws IoTDBConnectionException, StatementExecutionException { + final List<IMeasurementSchema> schemas = Collections.singletonList(chunk.getSchema()); + final boolean tableModel = deviceId.isTableModel(); + requireTableSessionIfNeeded(tableModel); + final TableTabletSchema tabletSchema = createTableTabletSchema(deviceId, schemas); + final List<TVList> lists = new ArrayList<>(); + lists.addAll(chunk.getSortedList()); + lists.add(chunk.getWorkingTVList()); + boolean replayed = false; + for (TVList list : lists) { + if (list == null || list.rowCount() == 0) { + continue; + } + if (!list.isSorted()) { + list.sort(); + } + for (int start = 0; start < list.rowCount(); start += SNAPSHOT_TABLET_ROW_LIMIT) { + final int end = Math.min(start + SNAPSHOT_TABLET_ROW_LIMIT, list.rowCount()); + final Tablet tablet = + buildNonAlignedTablet(deviceId, tabletSchema, schemas, list, start, end); + sendTablet(tablet, tableModel, false); + replayed = true; + } + } + return replayed; + } + + private boolean replayAlignedMemChunk( + final IDeviceID deviceId, final AlignedWritableMemChunk chunk) + throws IoTDBConnectionException, StatementExecutionException { + final boolean tableModel = deviceId.isTableModel(); + requireTableSessionIfNeeded(tableModel); + final List<IMeasurementSchema> schemas = chunk.getSchemaList(); + final TableTabletSchema tabletSchema = createTableTabletSchema(deviceId, schemas); + final List<AlignedTVList> lists = new ArrayList<>(); + lists.addAll(chunk.getSortedList()); + lists.add(chunk.getWorkingTVList()); + boolean replayed = false; + for (AlignedTVList list : lists) { + if (list == null || list.rowCount() == 0) { + continue; + } + if (!list.isSorted()) { + list.sort(); + } + final List<Integer> replayableRows = getReplayableAlignedRows(list); + for (int start = 0; start < replayableRows.size(); start += SNAPSHOT_TABLET_ROW_LIMIT) { + final int end = Math.min(start + SNAPSHOT_TABLET_ROW_LIMIT, replayableRows.size()); + final Tablet tablet = + buildAlignedTablet(deviceId, tabletSchema, schemas, list, replayableRows, start, end); + sendTablet(tablet, tableModel, true); + replayed = true; + } + } + return replayed; + } + + private static List<Integer> getReplayableAlignedRows(final AlignedTVList list) { + final List<Integer> replayableRows = new ArrayList<>(list.rowCount()); + for (int row = 0; row < list.rowCount(); row++) { + if (!list.isTimeDeleted(row)) { + replayableRows.add(row); + } + } + return replayableRows; + } + + private void requireTableSessionIfNeeded(final boolean tableModel) + throws StatementExecutionException { + if (tableModel && tableSession == null) { + throw new StatementExecutionException( + ImportWALMessages.EXCEPTION_TABLE_MODEL_WAL_ENTRIES_REQUIRE_DB_DATABASE_F7597726); + } + } + + private TableTabletSchema createTableTabletSchema( + final IDeviceID deviceId, final List<IMeasurementSchema> fieldSchemas) + throws IoTDBConnectionException, StatementExecutionException { + if (!deviceId.isTableModel()) { + return new TableTabletSchema(fieldSchemas, null, 0); + } + final String tableName = deviceId.getTableName(); + List<IMeasurementSchema> tagSchemas = tableTagSchemas.get(tableName); + if (tagSchemas == null) { + tagSchemas = new ArrayList<>(); + try (SessionDataSet dataSet = + tableSession.executeQueryStatement("DESCRIBE " + quoteIdentifier(tableName))) { + final SessionDataSet.DataIterator iterator = dataSet.iterator(); + while (iterator.next()) { + final String category = iterator.getString(3); + if ("TAG".equalsIgnoreCase(category)) { + tagSchemas.add( + new MeasurementSchema( + iterator.getString(1), TSDataType.valueOf(iterator.getString(2)))); + } + } + } + tableTagSchemas.put(tableName, tagSchemas); + } + final List<IMeasurementSchema> schemas = + new ArrayList<>(tagSchemas.size() + fieldSchemas.size()); + schemas.addAll(tagSchemas); + schemas.addAll(fieldSchemas); + final List<ColumnCategory> categories = new ArrayList<>(schemas.size()); + categories.addAll(Collections.nCopies(tagSchemas.size(), ColumnCategory.TAG)); + categories.addAll(Collections.nCopies(fieldSchemas.size(), ColumnCategory.FIELD)); + return new TableTabletSchema(schemas, categories, tagSchemas.size()); + } + + static String quoteIdentifier(final String identifier) { + return "\"" + identifier.replace("\"", "\"\"") + "\""; + } + + private void sendTablet(final Tablet tablet, final boolean tableModel, final boolean aligned) + throws IoTDBConnectionException, StatementExecutionException { + if (tableModel) { + tableSession.insertRelationalTablet(tablet); + } else if (aligned) { + treeSession.insertAlignedTablet(tablet); + } else { + treeSession.insertTablet(tablet); + } + } + + private static Tablet buildNonAlignedTablet( + final IDeviceID deviceId, + final TableTabletSchema tabletSchema, + final List<IMeasurementSchema> sourceSchemas, + final TVList list, + final int start, + final int end) { + final int rowCount = end - start; + final long[] times = new long[rowCount]; + final Object[] values = createValueArrays(tabletSchema.schemas, rowCount); + final BitMap[] bitMaps = new BitMap[tabletSchema.schemas.size()]; + final int fieldColumnIndex = tabletSchema.tagCount; + final TSDataType type = sourceSchemas.get(0).getType(); + for (int i = 0; i < rowCount; i++) { + final int scanIndex = start + i; + final int valueIndex = list.getValueIndex(scanIndex); + times[i] = list.getTime(scanIndex); + if (list.isNullValue(valueIndex)) { + if (bitMaps[fieldColumnIndex] == null) { + bitMaps[fieldColumnIndex] = new BitMap(rowCount); + } + bitMaps[fieldColumnIndex].mark(i); + } else { + putValue(values[fieldColumnIndex], i, type, list, scanIndex); + } + } + populateTableTags(deviceId, tabletSchema.categories, values, bitMaps, rowCount); + return tabletSchema.categories == null + ? new Tablet(deviceId.toString(), tabletSchema.schemas, times, values, bitMaps, rowCount) + : new Tablet( + deviceId.getTableName(), + tabletSchema.schemas, + tabletSchema.categories, + times, + values, + bitMaps, + rowCount); + } + + private static Tablet buildAlignedTablet( + final IDeviceID deviceId, + final TableTabletSchema tabletSchema, + final List<IMeasurementSchema> sourceSchemas, + final AlignedTVList list, + final List<Integer> replayableRows, + final int start, + final int end) { + final int rowCount = end - start; + final long[] times = new long[rowCount]; + final Object[] values = createValueArrays(tabletSchema.schemas, rowCount); + final BitMap[] bitMaps = new BitMap[tabletSchema.schemas.size()]; + final List<TSDataType> types = list.getTsDataTypes(); + for (int i = 0; i < rowCount; i++) { + final int scanIndex = replayableRows.get(start + i); + final int valueIndex = list.getValueIndex(scanIndex); + times[i] = list.getTime(scanIndex); + for (int c = 0; c < sourceSchemas.size(); c++) { + final int targetColumnIndex = tabletSchema.tagCount + c; + if (c >= types.size() || list.isNullValue(valueIndex, c)) { + if (bitMaps[targetColumnIndex] == null) { + bitMaps[targetColumnIndex] = new BitMap(rowCount); + } + bitMaps[targetColumnIndex].mark(i); + } else { + putValue(values[targetColumnIndex], i, types.get(c), list, valueIndex, c); + } + } + } + populateTableTags(deviceId, tabletSchema.categories, values, bitMaps, rowCount); + return tabletSchema.categories == null + ? new Tablet(deviceId.toString(), tabletSchema.schemas, times, values, bitMaps, rowCount) + : new Tablet( + deviceId.getTableName(), + tabletSchema.schemas, + tabletSchema.categories, + times, + values, + bitMaps, + rowCount); + } + + private static void populateTableTags( + final IDeviceID deviceId, + final List<ColumnCategory> columnCategories, + final Object[] values, + final BitMap[] bitMaps, + final int rowCount) { + if (columnCategories == null) { + return; + } + int tagSegmentIndex = 1; + for (int columnIndex = 0; columnIndex < columnCategories.size(); columnIndex++) { + if (columnCategories.get(columnIndex) != ColumnCategory.TAG) { + continue; + } + final Object segment = + tagSegmentIndex < deviceId.segmentNum() ? deviceId.segment(tagSegmentIndex) : null; + tagSegmentIndex++; + final Binary tagValue = + segment == null ? null : new Binary(segment.toString(), TSFileConfig.STRING_CHARSET); + for (int row = 0; row < rowCount; row++) { + if (tagValue == null) { + if (bitMaps[columnIndex] == null) { + bitMaps[columnIndex] = new BitMap(rowCount); + } + bitMaps[columnIndex].mark(row); + } else { + ((Binary[]) values[columnIndex])[row] = tagValue; + if (bitMaps[columnIndex] != null) { + bitMaps[columnIndex].unmark(row); + } + } + } + } + } + + private static Object createValueArray(final TSDataType type, final int rowCount) { + return switch (type) { + case BOOLEAN -> new boolean[rowCount]; + case INT32 -> new int[rowCount]; + case DATE -> new LocalDate[rowCount]; + case INT64, TIMESTAMP -> new long[rowCount]; + case FLOAT -> new float[rowCount]; + case DOUBLE -> new double[rowCount]; + case TEXT, STRING, BLOB, OBJECT -> new Binary[rowCount]; + case VECTOR, UNKNOWN -> throw unsupportedSnapshotDataType(type); + }; + } + + private static Object[] createValueArrays( + final List<IMeasurementSchema> schemas, final int rowCount) { + final Object[] values = new Object[schemas.size()]; + for (int column = 0; column < schemas.size(); column++) { + values[column] = createValueArray(schemas.get(column).getType(), rowCount); + } + return values; + } + + private static void putValue( + final Object target, + final int targetIndex, + final TSDataType type, + final TVList list, + final int sourceIndex) { + switch (type) { + case BOOLEAN -> ((boolean[]) target)[targetIndex] = list.getBoolean(sourceIndex); + case INT32 -> ((int[]) target)[targetIndex] = list.getInt(sourceIndex); + case DATE -> + ((LocalDate[]) target)[targetIndex] = + DateUtils.parseIntToLocalDate(list.getInt(sourceIndex)); + case INT64, TIMESTAMP -> ((long[]) target)[targetIndex] = list.getLong(sourceIndex); + case FLOAT -> ((float[]) target)[targetIndex] = list.getFloat(sourceIndex); + case DOUBLE -> ((double[]) target)[targetIndex] = list.getDouble(sourceIndex); + case TEXT, STRING, BLOB, OBJECT -> + ((Binary[]) target)[targetIndex] = list.getBinary(sourceIndex); + case VECTOR, UNKNOWN -> throw unsupportedSnapshotDataType(type); + } + } + + private static void putValue( + final Object target, + final int targetIndex, + final TSDataType type, + final AlignedTVList list, + final int sourceIndex, + final int columnIndex) { + switch (type) { + case BOOLEAN -> + ((boolean[]) target)[targetIndex] = + list.getBooleanByValueIndex(sourceIndex, columnIndex); + case INT32 -> + ((int[]) target)[targetIndex] = list.getIntByValueIndex(sourceIndex, columnIndex); + case DATE -> + ((LocalDate[]) target)[targetIndex] = + DateUtils.parseIntToLocalDate(list.getIntByValueIndex(sourceIndex, columnIndex)); + case INT64, TIMESTAMP -> + ((long[]) target)[targetIndex] = list.getLongByValueIndex(sourceIndex, columnIndex); + case FLOAT -> + ((float[]) target)[targetIndex] = list.getFloatByValueIndex(sourceIndex, columnIndex); + case DOUBLE -> + ((double[]) target)[targetIndex] = list.getDoubleByValueIndex(sourceIndex, columnIndex); + case TEXT, STRING, BLOB, OBJECT -> + ((Binary[]) target)[targetIndex] = list.getBinaryByValueIndex(sourceIndex, columnIndex); + case VECTOR, UNKNOWN -> throw unsupportedSnapshotDataType(type); + } + } + + private static IllegalArgumentException unsupportedSnapshotDataType(final TSDataType type) { + return new IllegalArgumentException( + String.format( + ImportWALMessages.EXCEPTION_UNSUPPORTED_SNAPSHOT_DATA_TYPE_ARG_7A32D312, type)); + } + + private static class TableTabletSchema { + private final List<IMeasurementSchema> schemas; + private final List<ColumnCategory> categories; + private final int tagCount; + + private TableTabletSchema( + final List<IMeasurementSchema> schemas, + final List<ColumnCategory> categories, + final int tagCount) { + this.schemas = schemas; + this.categories = categories; + this.tagCount = tagCount; + } + } + } + + static class ReplayStatistics { + private long replayedOperationCount; + private long skippedEntryCount; + private long totalBytes; + private long processedBytes; + private long completedFileCount; + private long elapsedNanos; + + long getReplayedOperationCount() { + return replayedOperationCount; + } + + long getSkippedEntryCount() { + return skippedEntryCount; + } + + long getTotalBytes() { + return totalBytes; + } + + long getCompletedFileCount() { + return completedFileCount; + } + + double getElapsedSeconds() { + return elapsedNanos / 1_000_000_000.0; + } + + double getProgressPercent() { + return totalBytes == 0 ? 100.0 : processedBytes * 100.0 / totalBytes; + } + + double getAverageRateMbPerSecond() { + final double elapsedSeconds = getElapsedSeconds(); + return elapsedSeconds <= 0 ? 0.0 : processedBytes / elapsedSeconds / (1024.0 * 1024.0); + } + } + + private static class WALReplayException extends IOException { + private WALReplayException(final String message, final Throwable cause) { + super(message, cause); + } + } +} 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 new file mode 100644 index 00000000000..787e06e1975 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/tools/ImportWALTest.java @@ -0,0 +1,506 @@ +/* + * 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.db.tools; + +import org.apache.iotdb.commons.path.MeasurementPath; +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.RelationalDeleteDataNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertTabletNode; +import org.apache.iotdb.db.storageengine.dataregion.memtable.IMemTable; +import org.apache.iotdb.db.storageengine.dataregion.memtable.PrimitiveMemTable; +import org.apache.iotdb.db.storageengine.dataregion.wal.WALTestUtils; +import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntry; +import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntryType; +import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALInfoEntry; +import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALSignalEntry; +import org.apache.iotdb.db.storageengine.dataregion.wal.io.ILogWriter; +import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALFileTest; +import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALWriter; +import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALByteBufferForTest; +import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALFileStatus; +import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALFileUtils; +import org.apache.iotdb.rpc.StatementExecutionException; +import org.apache.iotdb.session.Session; + +import org.apache.commons.cli.CommandLine; +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.file.metadata.StringArrayDeviceID; +import org.apache.tsfile.write.record.Tablet; +import org.apache.tsfile.write.schema.IMeasurementSchema; +import org.apache.tsfile.write.schema.MeasurementSchema; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import org.mockito.ArgumentCaptor; + +import java.io.ByteArrayOutputStream; +import java.io.File; +import java.io.IOException; +import java.io.PrintStream; +import java.nio.ByteBuffer; +import java.nio.file.Files; +import java.nio.file.Path; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; + +import static org.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +public class ImportWALTest { + + @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder(); + + /** + * Covers recursive directory discovery with WAL versions 2 and 10 in separate node folders. The + * result must ignore non-WAL files and preserve parent-folder then numeric-version order. + */ + @Test + public void testCollectWALFilesRecursivelyAndSortByVersion() throws IOException { + final Path source = temporaryFolder.newFolder("wal-root").toPath(); + final Path nodeA = Files.createDirectory(source.resolve("node-a")); + final Path nodeB = Files.createDirectory(source.resolve("node-b")); + final Path a10 = createWALFile(nodeA, 10); + final Path a2 = createWALFile(nodeA, 2); + final Path b1 = createWALFile(nodeB, 1); + Files.createFile(nodeA.resolve("ignore.txt")); + + final List<Path> files = ImportWAL.collectWALFiles(source); + + assertEquals( + Arrays.asList( + a2.toAbsolutePath().normalize(), + a10.toAbsolutePath().normalize(), + b1.toAbsolutePath().normalize()), + files); + } + + /** + * 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. + */ + @Test + public void testReplayWALFileReplaysInsertAndSkipsInternalEntry() throws Exception { + final File walFile = createWALFile(0); + final InsertRowNode rowNode = WALTestUtils.getInsertRowNode("root.sg.d1", 100); + writeWAL(walFile, new WALInfoEntry(1, rowNode), new WALSignalEntry(WALEntryType.CLOSE_SIGNAL)); + final Session treeSession = mock(Session.class); + + final ImportWAL.ReplayStatistics statistics = + ImportWAL.replayWALFiles( + Collections.singletonList(walFile.toPath()), + new ImportWAL.WALReplayer(treeSession, null, null)); + + assertEquals(1, statistics.getReplayedOperationCount()); + assertEquals(1, statistics.getSkippedEntryCount()); + final ArgumentCaptor<Tablet> tabletCaptor = ArgumentCaptor.forClass(Tablet.class); + verify(treeSession).insertTablet(tabletCaptor.capture()); + assertEquals("root.sg.d1", tabletCaptor.getValue().getDeviceId()); + assertEquals(100, tabletCaptor.getValue().getTimestamp(0)); + } + + @Test + public void testReplayReportsProgressAndFileStatistics() throws Exception { + final File walFile = createWALFile(0); + writeWAL( + walFile, + new WALInfoEntry(1, WALTestUtils.getInsertRowNode("root.sg.progress", 100)), + new WALSignalEntry(WALEntryType.CLOSE_SIGNAL)); + final ByteArrayOutputStream output = new ByteArrayOutputStream(); + + final ImportWAL.ReplayStatistics statistics = + ImportWAL.replayWALFiles( + Collections.singletonList(walFile.toPath()), + new ImportWAL.WALReplayer(mock(Session.class), null, null), + new PrintStream(output)); + + assertEquals(1, statistics.getCompletedFileCount()); + assertEquals(Files.size(walFile.toPath()), statistics.getTotalBytes()); + assertTrue(statistics.getElapsedSeconds() >= 0); + assertTrue(output.toString().contains("1/1")); + } + + /** + * Covers an active WAL containing one complete entry but no end marker or metadata. Replay must + * treat EOF at the entry boundary as clean and import the entry without waiting for writer close. + */ + @Test + public void testReplayActiveWALFileAtEntryBoundary() throws Exception { + final File walFile = createWALFile(0); + final InsertRowNode rowNode = WALTestUtils.getInsertRowNode("root.sg.active", 102); + final Session treeSession = mock(Session.class); + + try (WALWriter writer = new WALWriter(walFile)) { + writer.write(serializeWAL(new WALInfoEntry(1, rowNode))); + writer.force(); + + final ImportWAL.ReplayStatistics statistics = + ImportWAL.replayWALFiles( + Collections.singletonList(walFile.toPath()), + new ImportWAL.WALReplayer(treeSession, null, null)); + + assertEquals(1, statistics.getReplayedOperationCount()); + assertEquals(0, statistics.getSkippedEntryCount()); + verify(treeSession).insertTablet(any(Tablet.class)); + } + } + + /** + * Covers an aligned tree row. The replay must use the aligned Session API and must not fall back + * to the non-aligned Tablet API. + */ + @Test + public void testReplayAlignedTreeInsertUsesAlignedSessionAPI() throws Exception { + final InsertRowNode rowNode = WALTestUtils.getInsertRowNode("root.sg.aligned", 101); + rowNode.setAligned(true); + final Session treeSession = mock(Session.class); + + new ImportWAL.WALReplayer(treeSession, null, null).replay(new WALInfoEntry(1, rowNode)); + + verify(treeSession).insertAlignedTablet(any(Tablet.class)); + verify(treeSession, never()).insertTablet(any(Tablet.class)); + } + + /** + * Covers a table-model WAL tablet with an explicit target database. Replay must use the table + * Session and retain the source table name in the converted Tablet. + */ + @Test + public void testReplayTableInsertUsesRelationalSessionAPI() throws Exception { + final RelationalInsertTabletNode node = WALFileTest.getRelationalInsertTabletNode("table1"); + final Session treeSession = mock(Session.class); + final Session tableSession = mock(Session.class); + + new ImportWAL.WALReplayer(treeSession, tableSession, "db").replay(new WALInfoEntry(1, node)); + + final ArgumentCaptor<Tablet> tabletCaptor = ArgumentCaptor.forClass(Tablet.class); + verify(tableSession).insertRelationalTablet(tabletCaptor.capture()); + assertEquals("table1", tabletCaptor.getValue().getTableName()); + verify(treeSession, never()).insertTablet(any(Tablet.class)); + } + + /** + * Covers a table-model WAL insert without a target database. Replay must fail before issuing any + * write because WAL insert entries do not carry their source database name. + */ + @Test + public void testReplayTableInsertRequiresDatabase() throws Exception { + final RelationalInsertTabletNode node = WALFileTest.getRelationalInsertTabletNode("table1"); + final Session treeSession = mock(Session.class); + + assertThrows( + StatementExecutionException.class, + () -> new ImportWAL.WALReplayer(treeSession, null, null).replay(new WALInfoEntry(1, node))); + + verify(treeSession, never()).insertTablet(any(Tablet.class)); + } + + /** + * Covers a tree deletion with multiple paths and a bounded time range. Replay must pass the + * original paths and inclusive time bounds to Session.deleteData. + */ + @Test + public void testReplayTreeDeletePreservesPathsAndTimeRange() throws Exception { + final DeleteDataNode deleteNode = + new DeleteDataNode( + new PlanNodeId(""), + Arrays.asList( + new MeasurementPath("root.sg.d1.s1"), new MeasurementPath("root.sg.d2.*")), + 10, + 20); + final Session treeSession = mock(Session.class); + + new ImportWAL.WALReplayer(treeSession, null, null).replay(new WALInfoEntry(1, deleteNode)); + + verify(treeSession) + .deleteData(eq(Arrays.asList("root.sg.d1.s1", "root.sg.d2.*")), eq(10L), eq(20L)); + } + + /** 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); + + assertFalse(replayed); + } + + /** Covers an unsupported entry when the interactive user declines the skip prompt. */ + @Test + public void testReplayUnsupportedEntryFailsAfterDecliningSkip() { + final WALEntry entry = mockUnsupportedEntry(); + + assertThrows( + StatementExecutionException.class, + () -> + new ImportWAL.WALReplayer(mock(Session.class), null, null, ignored -> false) + .replay(entry)); + } + + /** Covers non-interactive execution, which must retain the original fail-fast behavior. */ + @Test + public void testReplayUnsupportedEntryFailsWithoutInteractiveInput() { + final WALEntry entry = mockUnsupportedEntry(); + + assertThrows( + StatementExecutionException.class, + () -> new ImportWAL.WALReplayer(mock(Session.class), null, null, 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)); + } + + /** Covers a non-aligned snapshot whose measurements have independent time axes. */ + @Test + public void testReplayNonAlignedMemTableSnapshotAsTablets() throws Exception { + final PrimitiveMemTable memTable = new PrimitiveMemTable("root.sg", "0"); + final List<IMeasurementSchema> schemas = + Arrays.asList( + new MeasurementSchema("s1", TSDataType.INT32), + new MeasurementSchema("s2", TSDataType.INT64)); + final StringArrayDeviceID deviceId = new StringArrayDeviceID("root.sg.d1"); + memTable.write(deviceId, schemas, 3, new Object[] {30, 300L}); + memTable.write(deviceId, schemas, 1, new Object[] {10, null}); + final Session treeSession = mock(Session.class); + + new ImportWAL.WALReplayer(treeSession, null, null).replay(new WALInfoEntry(1, memTable)); + + final ArgumentCaptor<Tablet> tabletCaptor = ArgumentCaptor.forClass(Tablet.class); + verify(treeSession, times(2)).insertTablet(tabletCaptor.capture()); + final Tablet s1Tablet = + tabletCaptor.getAllValues().stream() + .filter(tablet -> "s1".equals(tablet.getSchemas().get(0).getMeasurementName())) + .findFirst() + .orElseThrow(AssertionError::new); + assertEquals(2, s1Tablet.getRowSize()); + assertEquals(1, s1Tablet.getTimestamp(0)); + assertEquals(3, s1Tablet.getTimestamp(1)); + assertArrayEquals(new int[] {10, 30}, (int[]) s1Tablet.getValues()[0]); + } + + /** Covers an aligned snapshot with nulls and verifies the aligned Session API is used. */ + @Test + public void testReplayAlignedMemTableSnapshotPreservesNulls() throws Exception { + final PrimitiveMemTable memTable = new PrimitiveMemTable("root.sg", "0"); + final List<IMeasurementSchema> schemas = + Arrays.asList( + new MeasurementSchema("s1", TSDataType.INT32), + new MeasurementSchema("s2", TSDataType.INT64)); + final StringArrayDeviceID deviceId = new StringArrayDeviceID("root.sg.d1"); + memTable.writeAlignedRow(deviceId, schemas, 2, new Object[] {20, null}); + memTable.writeAlignedRow(deviceId, schemas, 1, new Object[] {10, 100L}); + final Session treeSession = mock(Session.class); + + new ImportWAL.WALReplayer(treeSession, null, null).replay(new WALInfoEntry(1, memTable)); + + final ArgumentCaptor<Tablet> tabletCaptor = ArgumentCaptor.forClass(Tablet.class); + verify(treeSession).insertAlignedTablet(tabletCaptor.capture()); + verify(treeSession, never()).insertTablet(any(Tablet.class)); + final Tablet tablet = tabletCaptor.getValue(); + assertEquals(2, tablet.getRowSize()); + assertEquals(1, tablet.getTimestamp(0)); + assertEquals(2, tablet.getTimestamp(1)); + assertArrayEquals(new int[] {10, 20}, (int[]) tablet.getValues()[0]); + assertTrue(tablet.getBitMaps()[1].isMarked(1)); + } + + /** Covers snapshot serialization and deserialization through a real WAL file. */ + @Test + public void testReplaySerializedMemTableSnapshot() throws Exception { + final PrimitiveMemTable memTable = new PrimitiveMemTable("root.sg", "0"); + memTable.write( + new StringArrayDeviceID("root.sg.d1"), + Collections.singletonList(new MeasurementSchema("s1", TSDataType.INT32)), + 7, + new Object[] {70}); + final File walFile = createWALFile(0); + writeWAL(walFile, new WALInfoEntry(1, memTable)); + final Session treeSession = mock(Session.class); + + final ImportWAL.ReplayStatistics statistics = + ImportWAL.replayWALFiles( + Collections.singletonList(walFile.toPath()), + new ImportWAL.WALReplayer(treeSession, null, null)); + + assertEquals(1, statistics.getReplayedOperationCount()); + final ArgumentCaptor<Tablet> tabletCaptor = ArgumentCaptor.forClass(Tablet.class); + verify(treeSession).insertTablet(tabletCaptor.capture()); + assertEquals(7, tabletCaptor.getValue().getTimestamp(0)); + assertEquals(70, ((int[]) tabletCaptor.getValue().getValues()[0])[0]); + } + + /** Covers a snapshot larger than the replay batch limit. */ + @Test + public void testReplayMemTableSnapshotSplitsLargeChunk() throws Exception { + final PrimitiveMemTable memTable = new PrimitiveMemTable("root.sg", "0"); + final List<IMeasurementSchema> schemas = + Collections.singletonList(new MeasurementSchema("s1", TSDataType.INT32)); + final StringArrayDeviceID deviceId = new StringArrayDeviceID("root.sg.d1"); + for (int i = 0; i < 1025; i++) { + memTable.write(deviceId, schemas, i, new Object[] {i}); + } + final Session treeSession = mock(Session.class); + + new ImportWAL.WALReplayer(treeSession, null, null).replay(new WALInfoEntry(1, memTable)); + + final ArgumentCaptor<Tablet> tabletCaptor = ArgumentCaptor.forClass(Tablet.class); + verify(treeSession, times(2)).insertTablet(tabletCaptor.capture()); + assertEquals(1024, tabletCaptor.getAllValues().get(0).getRowSize()); + assertEquals(1, tabletCaptor.getAllValues().get(1).getRowSize()); + } + + /** Covers a signal snapshot, which carries no user data and must be skipped. */ + @Test + public void testReplaySignalMemTableSnapshotIsSkipped() throws Exception { + final IMemTable signalMemTable = mock(IMemTable.class); + when(signalMemTable.isSignalMemTable()).thenReturn(true); + final Session treeSession = mock(Session.class); + + final boolean replayed = + new ImportWAL.WALReplayer(treeSession, null, null) + .replay(new WALInfoEntry(1, signalMemTable)); + + assertFalse(replayed); + verify(treeSession, never()).insertTablet(any(Tablet.class)); + verify(treeSession, never()).insertAlignedTablet(any(Tablet.class)); + } + + /** Covers a table-model snapshot without a target database. */ + @Test + public void testReplayTableMemTableSnapshotRequiresDatabase() throws Exception { + final PrimitiveMemTable memTable = new PrimitiveMemTable("db", "0"); + memTable.writeAlignedRow( + new StringArrayDeviceID("table1", "device1"), + Collections.singletonList(new MeasurementSchema("temperature", TSDataType.FLOAT)), + 1, + new Object[] {1.0F}); + final Session treeSession = mock(Session.class); + + assertThrows( + StatementExecutionException.class, + () -> + new ImportWAL.WALReplayer(treeSession, null, null) + .replay(new WALInfoEntry(1, memTable))); + + verify(treeSession, never()).insertAlignedTablet(any(Tablet.class)); + } + + /** Covers table-model identifier quoting, including an embedded double quote. */ + @Test + public void testQuoteTableIdentifierForDescribe() { + assertEquals("\"table\"", ImportWAL.WALReplayer.quoteIdentifier("table")); + assertEquals("\"table\"\"name\"", ImportWAL.WALReplayer.quoteIdentifier("table\"name")); + } + + /** + * Covers a truncated WAL that cannot yield a complete entry. The file-level replay must fail so + * callers cannot mistake a partial replay for success. + */ + @Test + public void testReplayWALFileFailsOnCorruption() throws Exception { + final File walFile = createWALFile(0); + Files.write(walFile.toPath(), new byte[] {WALEntryType.INSERT_ROW_NODE.getCode()}); + + final IOException exception = + assertThrows( + IOException.class, + () -> + ImportWAL.replayWALFiles( + Collections.singletonList(walFile.toPath()), + new ImportWAL.WALReplayer(mock(Session.class), null, null))); + + assertTrue(exception.getMessage().contains(walFile.getName())); + } + + @Test + public void testPasswordIsRequiredWhenInteractiveInputIsUnavailable() { + final CommandLine commandLine = mock(CommandLine.class); + when(commandLine.hasOption("password")).thenReturn(false); + + assertThrows(IllegalArgumentException.class, () -> ImportWAL.getPassword(commandLine, null)); + } + + @Test + public void testExplicitPasswordTakesPrecedence() { + final CommandLine commandLine = mock(CommandLine.class); + when(commandLine.hasOption("password")).thenReturn(true); + when(commandLine.getOptionValue("password")).thenReturn("secret"); + + assertEquals("secret", ImportWAL.getPassword(commandLine, null)); + } + + private Path createWALFile(final Path parent, final long version) throws IOException { + return Files.createFile( + parent.resolve( + WALFileUtils.getLogFileName(version, 0, WALFileStatus.CONTAINS_SEARCH_INDEX))); + } + + private File createWALFile(final long version) throws IOException { + return temporaryFolder.newFile( + WALFileUtils.getLogFileName(version, 0, WALFileStatus.CONTAINS_SEARCH_INDEX)); + } + + private static void writeWAL(final File walFile, final WALEntry... entries) throws IOException { + try (ILogWriter writer = new WALWriter(walFile)) { + writer.write(serializeWAL(entries)); + } + } + + private static ByteBuffer serializeWAL(final WALEntry... entries) { + int serializedSize = 0; + for (final WALEntry entry : entries) { + serializedSize += entry.serializedSize(); + } + final WALByteBufferForTest buffer = + new WALByteBufferForTest(ByteBuffer.allocate(serializedSize)); + for (final WALEntry entry : entries) { + entry.serialize(buffer); + } + return buffer.getBuffer(); + } + + private static WALEntry mockUnsupportedEntry() { + final WALEntry entry = mock(WALEntry.class); + when(entry.getType()).thenReturn(WALEntryType.RELATIONAL_DELETE_DATA_NODE); + when(entry.getValue()).thenReturn(mock(RelationalDeleteDataNode.class)); + return entry; + } +} diff --git a/scripts/tools/import-wal.sh b/scripts/tools/import-wal.sh new file mode 100644 index 00000000000..58bfe63c9ee --- /dev/null +++ b/scripts/tools/import-wal.sh @@ -0,0 +1,52 @@ +#!/bin/bash +# +# 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. +# + +if [ -z "${IOTDB_INCLUDE}" ]; then + : +elif [ -r "$IOTDB_INCLUDE" ]; then + . "$IOTDB_INCLUDE" +fi + +if [ -z "${IOTDB_HOME}" ]; then + export IOTDB_HOME="$(cd "$(dirname "$0")"/..; pwd)" +fi + +if [ -n "$JAVA_HOME" ]; then + for java in "$JAVA_HOME"/bin/amd64/java "$JAVA_HOME"/bin/java; do + if [ -x "$java" ]; then + JAVA="$java" + break + fi + done +else + JAVA=java +fi + +if [ -z "$JAVA" ]; then + echo "Unable to find java executable. Check JAVA_HOME and PATH environment variables." > /dev/stderr + exit 1 +fi + +JVM_OPTS="-Dsun.jnu.encoding=UTF-8 -Dfile.encoding=UTF-8" +CLASSPATH="${IOTDB_HOME}/lib/*" +MAIN_CLASS=org.apache.iotdb.db.tools.ImportWAL + +"$JAVA" $JVM_OPTS -DIOTDB_HOME="${IOTDB_HOME}" -cp "$CLASSPATH" "$MAIN_CLASS" "$@" +exit $? diff --git a/scripts/tools/windows/import-wal.bat b/scripts/tools/windows/import-wal.bat new file mode 100644 index 00000000000..dc69c440d81 --- /dev/null +++ b/scripts/tools/windows/import-wal.bat @@ -0,0 +1,43 @@ +@REM +@REM Licensed to the Apache Software Foundation (ASF) under one +@REM or more contributor license agreements. See the NOTICE file +@REM distributed with this work for additional information +@REM regarding copyright ownership. The ASF licenses this file +@REM to you under the Apache License, Version 2.0 (the +@REM "License"); you may not use this file except in compliance +@REM with the License. You may obtain a copy of the License at +@REM +@REM http://www.apache.org/licenses/LICENSE-2.0 +@REM +@REM Unless required by applicable law or agreed to in writing, +@REM software distributed under the License is distributed on an +@REM "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +@REM KIND, either express or implied. See the License for the +@REM specific language governing permissions and limitations +@REM under the License. +@REM + +@echo off +if "%OS%" == "Windows_NT" setlocal + +pushd %~dp0..\.. +if NOT DEFINED IOTDB_HOME set IOTDB_HOME=%CD% +popd + +if NOT DEFINED MAIN_CLASS set MAIN_CLASS=org.apache.iotdb.db.tools.ImportWAL +if NOT DEFINED JAVA_HOME goto :err + +set JAVA_OPTS=-ea^ + -DIOTDB_HOME="%IOTDB_HOME%" -Dsun.jnu.encoding=UTF-8 -Dfile.encoding=UTF-8 +set CLASSPATH="%IOTDB_HOME%\lib\*" + +"%JAVA_HOME%\bin\java" %JAVA_OPTS% -cp %CLASSPATH% %MAIN_CLASS% %* +set ret_code=%ERRORLEVEL% +goto finally + +:err +echo JAVA_HOME environment variable must be set! +set ret_code=1 + +:finally +ENDLOCAL & EXIT /B %ret_code%
