diff --git a/.gitignore b/.gitignore index b890cf0..dd5d6cb 100644 --- a/.gitignore +++ b/.gitignore @@ -12,6 +12,8 @@ # Package Files # *.jar +# 迁移工具发行包依赖这些内置插件完成 V2 插件元数据升级，必须随源码一起提交。 +!plugins/*.jar *.war *.nar *.ear diff --git a/README.md b/README.md index 31d029d..698c80e 100644 --- a/README.md +++ b/README.md @@ -1,17 +1,52 @@ # DataEase 数据迁移工具 -浏览器访问 `http://localhost:8080`，填写 DataEase 2.0 和 3.0 的 SSH、MySQL 及安装目录信息后执行迁移。 +浏览器访问 `http://localhost:9888`，填写 DataEase 2.0 和 3.0 的服务器地址、MySQL 及安装目录信息后执行完整迁移。远程服务器通过 SSH 操作，本地服务器直接操作本地目录。 迁移顺序为： -1. 将源端 `data/i18n`、`font`、`exportData`、`map`、`geo`、`appearance` `static-resource` 打包并复制至目标端的 `data` 目录，并将发行包的三个内置插件 JAR 复制至目标端 `data/plugin`。 -2. 通过 JAR 内置的 MySQL Connector/J 读取源库的结构、数据、视图、存储过程、函数、触发器及事件。 -3. 删除目标端 JDBC URL 指向的数据库，使用 `utf8mb4` 与 `utf8mb4_0900_ai_ci` 重建，并通过 JDBC 写入迁移内容。 +1. 将源端实际存在的 `data/i18n`、`font`、`exportData`、`map`、`geo`、`appearance`、`static-resource`、`excel` 打包并复制至目标端的 `data` 目录，并将发行包内置的 V3 插件 JAR 复制至目标端 `data/plugin`。启用同步日志开关时，还会迁移 `logs/sync-task/task-handler-log`。不存在的可选目录会跳过；所有候选数据目录都不存在时任务会中止，避免错误安装路径产生空迁移。 +2. 优先使用发行包中与当前平台匹配的 `mysql`、`mysqldump` 迁移数据库；未找到可执行工具时，自动回退到 JAR 内置的 MySQL Connector/J，迁移表结构、数据、视图、存储过程、函数、触发器及事件。 +3. 删除目标端 JDBC URL 指向的数据库，使用 `utf8mb4` 与 `utf8mb4_0900_ai_ci` 重建并写入迁移内容。 4. 在目标端数据库执行内置的 `upgrade.sql` 升级脚本。 -5. 按名称更新飞书多维表格、Apache Hive、达梦插件数据。源端存在其他插件时，任务日志会提示需升级的插件名称。 +5. 在事务中按模块或名称兼容更新飞书多维表格、Apache Hive、达梦以及同步管理 PostgreSQL 源/目标插件数据。源端存在其他插件时，任务日志会提示需升级的插件名称。 仅支持 MySQL JDBC URL，例如 `jdbc:mysql://127.0.0.1:3306/dataease`。运行 JAR 的机器必须可通过 JDBC URL 直连源端和目标端 MySQL。填写的数据库用户需具备源库读取定义和数据的权限，以及目标库 `DROP`、`CREATE`、写入权限和执行升级脚本中 `ALTER`、`UPDATE`、`INSERT`、`DELETE` 等语句的权限。 +源端和目标端安装目录必须填写对应服务器上的非根目录绝对路径；远程文件操作面向 Linux，本地文件操作支持 macOS/Linux。为兼容迁移工具回退到内置 JDBC 的情况，请预先创建一个可连接的全新空目标数据库；迁移开始后该数据库仍会被删除并重建。迁移期间应停止源 V2 的业务写入，并必须停止目标 V3 服务，避免文件快照、数据库数据或目标业务表被并发修改。 + +迁移任一阶段发生错误都会终止整个任务。页面日志会显示失败阶段、异常类型和底层根因，服务端日志会记录完整异常堆栈。同步任务参数为空、JSON 无效或缺少源/目标数据源对象时，日志还会逐条列出异常任务 ID、名称和具体原因，但不会输出可能包含数据库密码的任务参数原文。失败不会自动回滚已经完成的文件复制、目标库重建或 MySQL DDL，也不会修改 V2 源库；请先按日志修复 V2 数据，再使用全新目标数据库重新执行完整迁移，不要在失败目标库上继续补跑脚本。 + +## 本地迁移 + +把源端或目标端服务器地址填写为 `localhost`、`127.x`、`::1` 或运行迁移程序这台机器的网卡地址后，工具会直接归档、解压对应的本地安装目录并复制插件 JAR，不会建立 SSH 连接；SSH 端口、用户名和密码可以留空。远程地址仍会通过 SSH 执行相同操作，并要求填写有效的 SSH 配置。 + +源端和目标端会分别判断，因此支持本地到本地、本地到远程、远程到本地和远程到远程。连接方式不会改变迁移内容，四种组合都会依次执行服务文件迁移、数据库结构及数据复制、`upgrade.sql`、通用插件数据更新、同步管理专项数据转换和同步 PostgreSQL 插件数据更新。直接操作本地文件目前适用于 macOS/Linux，并要求本机提供 `/bin/sh` 和 `tar`。 + +## 同步任务日志复制开关 + +同步任务物理日志可能达到数百 MB 甚至更大，因此默认不复制。需要迁移时，在启动 JAR 时添加参数： + +```bash +java -jar dataease-migration-1.0.0.jar --migration.files.copy-sync-task-logs=true +``` + +也可以使用环境变量： + +```bash +MIGRATION_COPY_SYNC_TASK_LOGS=true java -jar dataease-migration-1.0.0.jar +``` + +开启后会把源端 `${安装目录}/logs/sync-task/task-handler-log` 单独归档，并合并解压至目标端同名目录，不会复制其他应用日志。日期目录和以 `per_sync_task_log.id` 命名的 `.log` 文件会保持不变；源端目录不存在时会记录日志并安全跳过。该配置是启动级参数，修改后需要重启迁移程序。 + +## 本机数据库测试 + +源库和目标库可以位于同一个本机 MySQL，但数据库名必须不同，例如： + +- 源库：`jdbc:mysql://127.0.0.1:3306/dataease_v2_test` +- 目标库：`jdbc:mysql://127.0.0.1:3306/dataease_v3_test` + +请先把待测数据导入源库并创建可连接的空目标库。目标库用户需要具备 `DROP`、`CREATE`、写入以及升级脚本所需的 `ALTER`、`UPDATE`、`INSERT`、`DELETE` 权限；执行测试时目标库仍会被删除并重建。若要验证 JDBC 批量复制优化，请确保本地 `tools/mysql/<平台>/bin` 中没有可用的 `mysql` 和 `mysqldump`，否则工具会优先走原生导出/导入路径。 + ## 自动选择数据库迁移工具 应用启动时按当前操作系统和 CPU 架构，在 `tools/mysql/<平台>/bin` 查找 `mysql` 和 `mysqldump`。两个工具均存在且可执行时，自动使用本地工具迁移；否则自动使用 JAR 内置 JDBC 迁移。 diff --git a/plugins/postgresql-backend-sink-3.0.0.jar b/plugins/postgresql-backend-sink-3.0.0.jar new file mode 100644 index 0000000..894388c Binary files /dev/null and b/plugins/postgresql-backend-sink-3.0.0.jar differ diff --git a/plugins/postgresql-backend-source-3.0.0.jar b/plugins/postgresql-backend-source-3.0.0.jar new file mode 100644 index 0000000..f908cf5 Binary files /dev/null and b/plugins/postgresql-backend-source-3.0.0.jar differ diff --git a/src/main/java/com/dataease/migration/model/ServerInfo.java b/src/main/java/com/dataease/migration/model/ServerInfo.java index 5f9db3e..01d3dae 100644 --- a/src/main/java/com/dataease/migration/model/ServerInfo.java +++ b/src/main/java/com/dataease/migration/model/ServerInfo.java @@ -1,15 +1,19 @@ package com.dataease.migration.model; -import jakarta.validation.constraints.Max; -import jakarta.validation.constraints.Min; import jakarta.validation.constraints.NotBlank; +/** + * DataEase 服务端文件位置及连接信息。 + * + *

host 和 installPath 对本地、远程迁移都必填；SSH 字段只在 host 指向远程机器时使用。 + * 因为 Bean Validation 无法根据 host 是否属于本机做条件校验，username、password、port + * 刻意不声明全局非空/范围约束，改由 MigrationService 在任务入队前按连接方式校验。

+ */ public record ServerInfo( @NotBlank(message = "服务器 IP 不能为空") String host, - @NotBlank(message = "服务器用户名不能为空") String username, - @NotBlank(message = "服务器密码不能为空") String password, - @Min(value = 1, message = "SSH 端口必须介于 1 到 65535") - @Max(value = 65535, message = "SSH 端口必须介于 1 到 65535") int port, + String username, + String password, + int port, @NotBlank(message = "DataEase 安装目录不能为空") String installPath ) { } diff --git a/src/main/java/com/dataease/migration/service/JdbcDatabaseMigrator.java b/src/main/java/com/dataease/migration/service/JdbcDatabaseMigrator.java index 97ac13b..7c882ee 100644 --- a/src/main/java/com/dataease/migration/service/JdbcDatabaseMigrator.java +++ b/src/main/java/com/dataease/migration/service/JdbcDatabaseMigrator.java @@ -13,20 +13,37 @@ import java.sql.Statement; import java.util.ArrayList; import java.util.List; +import java.util.Locale; +import java.util.Properties; +import java.util.concurrent.TimeUnit; +/** + * 没有可用 mysql/mysqldump 时使用的 JDBC 回退迁移器，负责复制 MySQL 结构和数据。 + * 大表数据采用流式读取、批量改写和分段提交，避免一次性加载全表或逐行网络往返。 + */ @Component public class JdbcDatabaseMigrator implements DatabaseMigrator { + /** + * 单批行数需要同时兼顾吞吐和内存：Connector/J 会将一批改写成多值 INSERT，而同步日志表 + * 可能含有较大的 TEXT/LONGTEXT，批次过大会显著抬高迁移进程及 MySQL 的瞬时内存占用。 + */ private static final int BATCH_SIZE = 500; + /** + * 每 20 批提交一次，减少逐批提交的 fsync 开销；用 BATCH_SIZE 计算可保证提交前没有待执行批次。 + */ + private static final int COMMIT_INTERVAL = BATCH_SIZE * 20; @Override public void migrate(DatabaseInfo source, DatabaseInfo target, MigrationJob job) throws SQLException { DatabaseConnection sourceConnection = DatabaseConnection.fromJdbcUrl(source.jdbcUrl()); DatabaseConnection targetConnection = DatabaseConnection.fromJdbcUrl(target.jdbcUrl()); - try (Connection sourceDb = connect(source); Connection targetDb = connect(target)) { + try (Connection sourceDb = connect(source, false); Connection targetDb = connect(target, true)) { sourceDb.setAutoCommit(false); recreateDatabase(targetDb, targetConnection, job); targetDb.setAutoCommit(false); execute(targetDb, "SET FOREIGN_KEY_CHECKS = 0"); + job.log("JDBC 批量写入已启用：每批 " + BATCH_SIZE + " 行，每 " + + COMMIT_INTERVAL + " 行提交一次。"); try { List tables = findObjects(sourceDb, sourceConnection.database(), "BASE TABLE"); job.log("正在迁移 " + tables.size() + " 个数据表结构。"); @@ -55,8 +72,24 @@ public void migrate(DatabaseInfo source, DatabaseInfo target, MigrationJob job) } } - private Connection connect(DatabaseInfo database) throws SQLException { - return DriverManager.getConnection(database.jdbcUrl(), database.username(), database.password()); + private Connection connect(DatabaseInfo database, boolean optimizeBatchWrites) throws SQLException { + return DriverManager.getConnection(database.jdbcUrl(), connectionProperties(database, optimizeBatchWrites)); + } + + /** + * 只为目标连接启用批量改写。源连接负责流式读取，无需写入优化；目标连接关闭服务端预编译后， + * Connector/J 才能稳定地把 executeBatch 改写为一条多值 INSERT。单独传 Properties 还会覆盖 + * JDBC URL 中的同名配置，避免页面输入的参数意外关闭迁移工具的批量策略。 + */ + static Properties connectionProperties(DatabaseInfo database, boolean optimizeBatchWrites) { + Properties properties = new Properties(); + properties.setProperty("user", database.username()); + properties.setProperty("password", database.password()); + if (optimizeBatchWrites) { + properties.setProperty("rewriteBatchedStatements", "true"); + properties.setProperty("useServerPrepStmts", "false"); + } + return properties; } private void recreateDatabase(Connection target, DatabaseConnection connection, MigrationJob job) throws SQLException { @@ -99,31 +132,60 @@ private void copyTableData(Connection source, Connection target, String sourceDa String insert = "INSERT INTO " + targetTable + " (" + quotedColumns + ") VALUES (" + placeholders + ")"; job.log("正在迁移数据表：" + table + "。"); + long startedNanos = System.nanoTime(); try (Statement read = source.createStatement(ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY); PreparedStatement write = target.prepareStatement(insert)) { read.setFetchSize(Integer.MIN_VALUE); try (ResultSet rows = read.executeQuery(select)) { int rowCount = 0; + int pendingBatchRows = 0; while (rows.next()) { for (int index = 0; index < columns.size(); index++) { write.setObject(index + 1, rows.getObject(index + 1)); } write.addBatch(); rowCount++; - if (rowCount % BATCH_SIZE == 0) { + pendingBatchRows++; + if (pendingBatchRows == BATCH_SIZE) { write.executeBatch(); + write.clearBatch(); + pendingBatchRows = 0; + } + if (rowCount % COMMIT_INTERVAL == 0) { + // COMMIT_INTERVAL 是 BATCH_SIZE 的整数倍，此时所有 addBatch 内容均已落库。 target.commit(); + job.log("数据表 " + table + " 已迁移 " + rowCount + " 行，平均 " + + rowsPerSecond(rowCount, startedNanos) + " 行/秒。"); } } - if (rowCount % BATCH_SIZE != 0) { + if (pendingBatchRows > 0) { write.executeBatch(); + write.clearBatch(); + } + // 整万行已在循环内提交；这里只提交最后不足一个提交区间的尾批，避免空提交。 + if (rowCount % COMMIT_INTERVAL != 0) { target.commit(); } - job.log("数据表 " + table + " 迁移完成，共 " + rowCount + " 行。"); + job.log("数据表 " + table + " 迁移完成，共 " + rowCount + " 行，耗时 " + + formatElapsed(startedNanos) + "，平均 " + rowsPerSecond(rowCount, startedNanos) + + " 行/秒。"); } } } + private long rowsPerSecond(int rowCount, long startedNanos) { + long elapsedNanos = Math.max(1, System.nanoTime() - startedNanos); + return Math.round(rowCount * 1_000_000_000.0 / elapsedNanos); + } + + private String formatElapsed(long startedNanos) { + long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startedNanos); + if (elapsedMillis < 1_000) { + return elapsedMillis + " 毫秒"; + } + return String.format(Locale.ROOT, "%.2f 秒", elapsedMillis / 1_000.0); + } + private List migratableColumns(Connection connection, String database, String table) throws SQLException { List columns = new ArrayList<>(); try (Statement statement = connection.createStatement(); diff --git a/src/main/java/com/dataease/migration/service/MigrationJob.java b/src/main/java/com/dataease/migration/service/MigrationJob.java index a969a9a..dcae5d8 100644 --- a/src/main/java/com/dataease/migration/service/MigrationJob.java +++ b/src/main/java/com/dataease/migration/service/MigrationJob.java @@ -17,6 +17,7 @@ public final class MigrationJob { private final List subscribers = new CopyOnWriteArrayList<>(); private final Instant createdAt = Instant.now(); private boolean completed; + private boolean succeeded; public void log(String message) { String line = "[" + Instant.now() + "] " + message; @@ -40,12 +41,13 @@ public void subscribe(SseEmitter emitter) { complete(emitter); } - public void complete() { + public void complete(boolean succeeded) { List activeSubscribers; synchronized (this) { if (completed) { return; } + this.succeeded = succeeded; completed = true; activeSubscribers = new ArrayList<>(subscribers); } @@ -66,7 +68,10 @@ private void send(SseEmitter emitter, String name, String data) { private void complete(SseEmitter emitter) { try { - emitter.send(SseEmitter.event().name("complete").data("迁移任务结束")); + String message = succeeded + ? "迁移任务成功完成" + : "迁移任务已失败，请根据上方异常明细处理后，使用全新目标库重新迁移"; + emitter.send(SseEmitter.event().name("complete").data(message)); } catch (IOException ignored) { // The browser may have disconnected before the final event. } finally { diff --git a/src/main/java/com/dataease/migration/service/MigrationService.java b/src/main/java/com/dataease/migration/service/MigrationService.java index 54869b8..154a772 100644 --- a/src/main/java/com/dataease/migration/service/MigrationService.java +++ b/src/main/java/com/dataease/migration/service/MigrationService.java @@ -3,42 +3,86 @@ import com.dataease.migration.model.DatabaseConnection; import com.dataease.migration.model.MigrationRequest; import com.dataease.migration.model.ServerInfo; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; +import java.net.InetAddress; +import java.net.NetworkInterface; +import java.nio.charset.StandardCharsets; import java.nio.file.Files; import java.nio.file.Path; +import java.nio.file.StandardCopyOption; +import java.util.Arrays; +import java.util.LinkedHashSet; +import java.util.Locale; import java.util.Map; +import java.util.Set; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.Executor; +import java.util.concurrent.TimeUnit; +/** + * 编排一次完整的 V2 到 V3 迁移。 + * + *

源端和目标端各自判断为本地文件访问或 SSH 访问，文件传输完成后再依次执行数据库复制、 + * 通用升级、插件升级和同步管理专项升级。这个顺序不能交换：后续步骤依赖前一步创建的 V3 表结构， + * 且任一步失败都应阻止尚未开始的后续转换。

+ */ @Service public class MigrationService { + private static final Logger LOGGER = LoggerFactory.getLogger(MigrationService.class); private static final String FILE_ARCHIVE_PREFIX = "/tmp/dataease-files-"; - private static final String[] DATA_DIRECTORIES = {"i18n", "font", "exportData", "map", "geo", "appearance", "static-resource"}; + private static final String SYNC_TASK_LOG_DIRECTORY = "logs/sync-task/task-handler-log"; + /** + * V2 持久化数据目录。不同版本或部署方式可能缺少其中部分目录，所以打包时按实际存在情况选择； + * excel 必须保留，否则 Excel 数据集的原始上传文件会在数据库迁移成功后丢失。 + */ + private static final String[] DATA_DIRECTORIES = { + "i18n", "font", "exportData", "map", "geo", "appearance", "static-resource", "excel" + }; private final SshCommandExecutor ssh; private final DatabaseMigrationSelector databaseMigrator; private final TargetDatabaseUpgradeService targetDatabaseUpgradeService; private final PluginMigrationService pluginMigrationService; + private final SyncManagementMigrationService syncManagementMigrationService; private final Executor migrationExecutor; + private final boolean copySyncTaskLogs; private final Map jobs = new ConcurrentHashMap<>(); public MigrationService(SshCommandExecutor ssh, DatabaseMigrationSelector databaseMigrator, TargetDatabaseUpgradeService targetDatabaseUpgradeService, PluginMigrationService pluginMigrationService, - @Qualifier("migrationExecutor") Executor migrationExecutor) { + SyncManagementMigrationService syncManagementMigrationService, + @Qualifier("migrationExecutor") Executor migrationExecutor, + @Value("${migration.files.copy-sync-task-logs:false}") boolean copySyncTaskLogs) { this.ssh = ssh; this.databaseMigrator = databaseMigrator; this.targetDatabaseUpgradeService = targetDatabaseUpgradeService; this.pluginMigrationService = pluginMigrationService; + this.syncManagementMigrationService = syncManagementMigrationService; this.migrationExecutor = migrationExecutor; + this.copySyncTaskLogs = copySyncTaskLogs; + LOGGER.info("同步任务物理日志复制已{}；启动参数：--migration.files.copy-sync-task-logs={}", + copySyncTaskLogs ? "启用" : "关闭", copySyncTaskLogs); } public String start(MigrationRequest request) { - DatabaseConnection.fromJdbcUrl(request.sourceDatabase().jdbcUrl()); - DatabaseConnection.fromJdbcUrl(request.targetDatabase().jdbcUrl()); + if (request.sourceServer() == null || request.targetServer() == null) { + throw new IllegalArgumentException("迁移必须填写源端和目标端服务器配置"); + } + validateServer(request.sourceServer(), "DataEase 2.0"); + validateServer(request.targetServer(), "DataEase 3.0"); + DatabaseConnection sourceDatabase = DatabaseConnection.fromJdbcUrl(request.sourceDatabase().jdbcUrl()); + DatabaseConnection targetDatabase = DatabaseConnection.fromJdbcUrl(request.targetDatabase().jdbcUrl()); + // 目标库会在数据库阶段先被删除；任务入队前拒绝明确相同的库，避免把源数据一起删除。 + if (sameDatabase(sourceDatabase, targetDatabase)) { + throw new IllegalArgumentException("源数据库和目标数据库不能相同，目标库会被删除并重建"); + } String id = UUID.randomUUID().toString(); MigrationJob job = new MigrationJob(); @@ -58,56 +102,318 @@ public MigrationJob getJob(String id) { private void runMigration(MigrationRequest request, MigrationJob job, String id) { Path localFiles = null; String remoteFiles = FILE_ARCHIVE_PREFIX + id + ".tar.gz"; + String currentStage = "初始化迁移任务"; + boolean succeeded = false; + long taskStartedNanos = System.nanoTime(); + long stageStartedNanos; try { - job.log("开始迁移任务。"); + job.log("开始完整迁移任务；源端和目标端将分别自动选择本地文件访问或 SSH 访问。"); localFiles = Files.createTempFile("dataease-files-", ".tar.gz"); + currentStage = "文件迁移"; + stageStartedNanos = System.nanoTime(); migrateFiles(request.sourceServer(), request.targetServer(), remoteFiles, localFiles, job); + logElapsed(job, currentStage, stageStartedNanos); + + currentStage = "数据库迁移"; + stageStartedNanos = System.nanoTime(); migrateDatabase(request, job); + logElapsed(job, currentStage, stageStartedNanos); + // 先完成通用 V3 表结构升级，再写入通用插件，最后执行依赖 V3/JPA 字段的同步管理专项迁移。 + currentStage = "V3 通用数据库结构升级"; + stageStartedNanos = System.nanoTime(); targetDatabaseUpgradeService.execute(request.targetDatabase(), job); - pluginMigrationService.updatePlugins(request.targetDatabase(), job); - job.log("迁移完成。"); + logElapsed(job, currentStage, stageStartedNanos); + currentStage = "通用插件迁移"; + stageStartedNanos = System.nanoTime(); + String targetInstallPath = request.targetServer().installPath(); + pluginMigrationService.updatePlugins(request.targetDatabase(), targetInstallPath, job); + logElapsed(job, currentStage, stageStartedNanos); + currentStage = "同步管理专项迁移"; + stageStartedNanos = System.nanoTime(); + syncManagementMigrationService.execute(request.targetDatabase(), targetInstallPath, job); + logElapsed(job, currentStage, stageStartedNanos); + job.log("完整迁移完成，总耗时：" + formatElapsed(taskStartedNanos) + "。"); + succeeded = true; } catch (Exception e) { - job.log("迁移失败：" + safeMessage(e)); + // 页面展示失败阶段和根因链，服务端日志保留完整堆栈，便于实施和研发分别定位。 + job.log("迁移失败，阶段：" + currentStage + "；原因：" + detailedMessage(e)); + LOGGER.error("迁移任务 {} 在阶段 [{}] 失败", id, currentStage, e); } finally { cleanup(request.sourceServer(), remoteFiles, job); cleanup(request.targetServer(), remoteFiles, job); deleteQuietly(localFiles); - job.complete(); + job.complete(succeeded); } } + /** + * 文件迁移固定经过迁移程序所在机器的临时归档，从而用同一流程覆盖本地→本地、本地→远程、 + * 远程→本地和远程→远程四种组合；只有文件访问方式不同，迁移内容保持一致。 + */ private void migrateFiles(ServerInfo source, ServerInfo target, String remoteArchive, Path localArchive, MigrationJob job) throws Exception { - job.log("开始迁移文件：i18n、font、exportData、map、geo、appearance static-resource 及插件。"); + job.log("开始迁移文件：i18n、font、exportData、map、geo、appearance、static-resource、excel 及插件。"); String sourceDataDirectory = source.installPath() + "/data"; - String directories = String.join(" ", DATA_DIRECTORIES); - ssh.execute(source, "tar -C " + ShellEscaper.quote(sourceDataDirectory) + " -czf " - + ShellEscaper.quote(remoteArchive) + " " + directories, job); - job.log("正在从 DataEase 2.0 服务器下载文件归档。"); - ssh.download(source, remoteArchive, localArchive); - job.log("正在上传文件归档到 DataEase 3.0 服务器。"); - ssh.upload(target, localArchive, remoteArchive); - ssh.execute(target, "mkdir -p " + ShellEscaper.quote(target.installPath() + "/data") - + " && tar -C " + ShellEscaper.quote(target.installPath() + "/data") - + " -xzf " + ShellEscaper.quote(remoteArchive), job); - String targetPluginDirectory = target.installPath() + "/data/plugin"; - ssh.execute(target, "mkdir -p " + ShellEscaper.quote(targetPluginDirectory), job); - for (Path pluginJar : pluginMigrationService.pluginJars()) { - String targetPlugin = targetPluginDirectory + "/" + pluginJar.getFileName(); - job.log("正在上传插件：" + pluginJar.getFileName() + "。"); - ssh.upload(target, pluginJar, targetPlugin); + boolean sourceLocal = isLocalHost(source.host()); + boolean targetLocal = isLocalHost(target.host()); + long stepStartedNanos = System.nanoTime(); + if (sourceLocal) { + job.log("源端地址 " + source.host() + " 属于本机，直接读取本地安装目录，不建立 SSH 连接。"); + executeLocal(buildFileArchiveCommand(sourceDataDirectory, localArchive.toString()), job); + job.log("本地文件归档完成，大小：" + formatBytes(Files.size(localArchive)) + + "，耗时：" + formatElapsed(stepStartedNanos) + "。"); + } else { + ssh.execute(source, buildFileArchiveCommand(sourceDataDirectory, remoteArchive), job); + logElapsed(job, "源端文件归档", stepStartedNanos); + job.log("正在从 DataEase 2.0 服务器下载文件归档。"); + stepStartedNanos = System.nanoTime(); + ssh.download(source, remoteArchive, localArchive); + job.log("文件归档下载完成，大小：" + formatBytes(Files.size(localArchive)) + + "，耗时：" + formatElapsed(stepStartedNanos) + "。"); + } + + if (targetLocal) { + job.log("目标端地址 " + target.host() + " 属于本机，直接写入本地安装目录，不建立 SSH 连接。"); + stepStartedNanos = System.nanoTime(); + extractArchiveLocally(target, localArchive, job); + logElapsed(job, "目标端本地文件解压", stepStartedNanos); + stepStartedNanos = System.nanoTime(); + copyPluginsLocally(target, job); + logElapsed(job, "本地插件文件复制", stepStartedNanos); + } else { + job.log("正在上传文件归档到 DataEase 3.0 服务器。"); + stepStartedNanos = System.nanoTime(); + ssh.upload(target, localArchive, remoteArchive); + logElapsed(job, "目标端文件归档上传", stepStartedNanos); + stepStartedNanos = System.nanoTime(); + ssh.execute(target, "mkdir -p " + ShellEscaper.quote(target.installPath() + "/data") + + " && tar -C " + ShellEscaper.quote(target.installPath() + "/data") + + " -xzf " + ShellEscaper.quote(remoteArchive), job); + logElapsed(job, "目标端文件解压", stepStartedNanos); + String targetPluginDirectory = target.installPath() + "/data/plugin"; + ssh.execute(target, "mkdir -p " + ShellEscaper.quote(targetPluginDirectory), job); + stepStartedNanos = System.nanoTime(); + for (Path pluginJar : pluginMigrationService.pluginJars()) { + String targetPlugin = targetPluginDirectory + "/" + pluginJar.getFileName(); + job.log("正在上传插件：" + pluginJar.getFileName() + "。"); + ssh.upload(target, pluginJar, targetPlugin); + } + logElapsed(job, "插件文件上传", stepStartedNanos); + } + if (copySyncTaskLogs) { + migrateSyncTaskLogs(source, target, remoteArchive, localArchive, sourceLocal, targetLocal, job); + } else { + job.log("未启用同步任务日志复制，跳过 " + SYNC_TASK_LOG_DIRECTORY + + "；可在启动时添加 --migration.files.copy-sync-task-logs=true 开启。"); } job.log("文件迁移完成。"); } + /** + * 同步任务物理日志可能远大于业务附件，因此默认不迁移，并在开关启用时使用独立归档步骤。 + * 此时主数据归档已经完成解压，可以安全复用临时文件路径，避免同时保留两份大归档占满磁盘。 + */ + private void migrateSyncTaskLogs(ServerInfo source, ServerInfo target, String remoteArchive, Path localArchive, + boolean sourceLocal, boolean targetLocal, MigrationJob job) throws Exception { + job.log("开始迁移同步任务日志：" + SYNC_TASK_LOG_DIRECTORY + "。"); + long stepStartedNanos = System.nanoTime(); + String sourceArchive = sourceLocal ? localArchive.toString() : remoteArchive; + String archiveCommand = buildSyncTaskLogArchiveCommand(source.installPath(), sourceArchive); + if (sourceLocal) { + executeLocal(archiveCommand, job); + } else { + ssh.execute(source, archiveCommand, job); + ssh.download(source, remoteArchive, localArchive); + } + job.log("同步任务日志归档准备完成，大小：" + formatBytes(Files.size(localArchive)) + + "，耗时：" + formatElapsed(stepStartedNanos) + "。"); + + String targetLogParent = target.installPath() + "/logs/sync-task"; + stepStartedNanos = System.nanoTime(); + if (targetLocal) { + executeLocal("mkdir -p " + ShellEscaper.quote(targetLogParent) + + " && tar -C " + ShellEscaper.quote(targetLogParent) + + " -xzf " + ShellEscaper.quote(localArchive.toString()), job); + } else { + ssh.upload(target, localArchive, remoteArchive); + ssh.execute(target, "mkdir -p " + ShellEscaper.quote(targetLogParent) + + " && tar -C " + ShellEscaper.quote(targetLogParent) + + " -xzf " + ShellEscaper.quote(remoteArchive), job); + } + logElapsed(job, "同步任务日志目标端合并", stepStartedNanos); + job.log("同步任务日志迁移完成。"); + } + + private void extractArchiveLocally(ServerInfo target, Path localArchive, MigrationJob job) throws Exception { + String targetDataDirectory = target.installPath() + "/data"; + executeLocal("mkdir -p " + ShellEscaper.quote(targetDataDirectory) + + " && tar -C " + ShellEscaper.quote(targetDataDirectory) + + " -xzf " + ShellEscaper.quote(localArchive.toString()), job); + } + + private void copyPluginsLocally(ServerInfo target, MigrationJob job) throws Exception { + Path targetPluginDirectory = Path.of(target.installPath(), "data", "plugin"); + Files.createDirectories(targetPluginDirectory); + for (Path pluginJar : pluginMigrationService.pluginJars()) { + Path targetPlugin = targetPluginDirectory.resolve(pluginJar.getFileName()); + job.log("正在复制本地插件：" + pluginJar.getFileName() + "。"); + Files.copy(pluginJar, targetPlugin, StandardCopyOption.REPLACE_EXISTING); + } + } + + /** + * 本地分支复用与 SSH 分支相同的、已完成 shell 引用的归档命令，并把输出写入任务日志。 + * 显式使用 /bin/sh 是为了让命令语法在支持的 macOS/Linux 环境中保持一致。 + */ + private void executeLocal(String command, MigrationJob job) throws Exception { + Process process = new ProcessBuilder("/bin/sh", "-c", command) + .redirectErrorStream(true) + .start(); + try (var reader = process.inputReader(StandardCharsets.UTF_8)) { + String line; + while ((line = reader.readLine()) != null) { + job.log(line); + } + } + int exitCode = process.waitFor(); + if (exitCode != 0) { + throw new IllegalStateException("本地文件操作失败，退出码：" + exitCode); + } + } + + /** + * 构造源端归档命令。使用 shell 位置参数只把真实存在的目录传给 tar： + * 一方面兼容没有 i18n、appearance 等可选目录的 V2 环境，另一方面所有路径都经过引用处理， + * 避免安装路径包含空格或特殊字符时被错误拆分。如果所有候选目录都不存在，则判定源安装目录 + * 无效或不完整并中止，不能用空归档伪装成文件迁移成功。 + */ + static String buildFileArchiveCommand(String sourceDataDirectory, String remoteArchive) { + String directories = String.join(" ", Arrays.stream(DATA_DIRECTORIES) + .map(ShellEscaper::quote) + .toList()); + return "cd " + ShellEscaper.quote(sourceDataDirectory) + " || exit 1; " + + "set --; " + + "for directory in " + directories + "; do " + + "if [ -d \"$directory\" ]; then set -- \"$@\" \"$directory\"; " + + "else echo \"跳过不存在的数据目录：$directory\"; fi; " + + "done; " + + "if [ \"$#\" -eq 0 ]; then " + + "echo \"未找到任何可迁移的 DataEase 2.0 数据目录\" >&2; exit 1; fi; " + + "tar -czf " + ShellEscaper.quote(remoteArchive) + " \"$@\"" + + " && du -h " + ShellEscaper.quote(remoteArchive); + } + + static String buildSyncTaskLogArchiveCommand(String sourceInstallPath, String archive) { + String sourceLogParent = sourceInstallPath + "/logs/sync-task"; + String sourceLogDirectory = sourceInstallPath + "/" + SYNC_TASK_LOG_DIRECTORY; + return "if [ -d " + ShellEscaper.quote(sourceLogDirectory) + " ]; then " + + "tar -C " + ShellEscaper.quote(sourceLogParent) + " -czf " + ShellEscaper.quote(archive) + + " task-handler-log; " + + "else echo " + ShellEscaper.quote("跳过不存在的同步任务日志目录：" + sourceLogDirectory) + "; " + + "tar -czf " + ShellEscaper.quote(archive) + " -T /dev/null; fi" + + " && du -h " + ShellEscaper.quote(archive); + } + + /** + * SSH 字段采用条件校验：本地地址不建立 SSH 连接，可留空；远程地址必须提供完整连接信息。 + * 安装目录在两种模式下都会进入 shell 命令，因此无论本地或远程都先执行同一套路径校验。 + */ + private static void validateServer(ServerInfo server, String productName) { + if (server.host() == null || server.host().isBlank()) { + throw new IllegalArgumentException(productName + " 服务器 IP 不能为空"); + } + validateLinuxInstallPath(server.installPath(), productName); + if (isLocalHost(server.host())) { + return; + } + if (server.port() < 1 || server.port() > 65535) { + throw new IllegalArgumentException(productName + " SSH 端口必须介于 1 到 65535"); + } + if (server.username() == null || server.username().isBlank()) { + throw new IllegalArgumentException(productName + " 远程地址必须填写 SSH 用户名"); + } + if (server.password() == null || server.password().isBlank()) { + throw new IllegalArgumentException(productName + " 远程地址必须填写 SSH 密码"); + } + } + + /** + * localhost、环回地址以及绑定在当前机器网卡上的地址都视为本地地址。域名解析失败时按远程地址处理， + * 后续会按远程服务器规则校验 SSH 配置，避免误将不可识别的主机当成本地文件系统。 + */ + static boolean isLocalHost(String host) { + if (host == null || host.isBlank()) { + return false; + } + String normalized = host.trim(); + if (normalized.startsWith("[") && normalized.endsWith("]")) { + normalized = normalized.substring(1, normalized.length() - 1); + } + if ("localhost".equalsIgnoreCase(normalized) || "localhost.".equalsIgnoreCase(normalized)) { + return true; + } + try { + for (InetAddress address : InetAddress.getAllByName(normalized)) { + if (address.isLoopbackAddress() + || (!address.isAnyLocalAddress() && NetworkInterface.getByInetAddress(address) != null)) { + return true; + } + } + } catch (Exception ignored) { + // 无法解析或读取本机网卡时按远程地址处理。 + } + return false; + } + + /** + * 比较源/目标是否明确指向同一库。除主机名文本比较外还比较解析后的地址，避免 localhost 与 + * 127.0.0.1 这类别名绕过删除保护；解析失败时保守退回文本比较，实际连接错误仍由数据库阶段报告。 + */ + static boolean sameDatabase(DatabaseConnection source, DatabaseConnection target) { + if (source.port() != target.port() || !source.database().equals(target.database())) { + return false; + } + if (source.host().equalsIgnoreCase(target.host())) { + return true; + } + try { + InetAddress[] sourceAddresses = InetAddress.getAllByName(source.host()); + InetAddress[] targetAddresses = InetAddress.getAllByName(target.host()); + return Arrays.stream(sourceAddresses) + .anyMatch(sourceAddress -> Arrays.stream(targetAddresses).anyMatch(sourceAddress::equals)); + } catch (Exception ignored) { + return false; + } + } + + /** + * 校验 DataEase 安装目录使用无歧义的 Linux/macOS 绝对路径。 + */ + private static void validateLinuxInstallPath(String installPath, String productName) { + if (installPath == null || installPath.isBlank()) { + throw new IllegalArgumentException(productName + " 安装目录不能为空"); + } + boolean containsRelativeSegment = Arrays.stream(installPath.split("/", -1)) + .anyMatch(segment -> ".".equals(segment) || "..".equals(segment)); + boolean containsControlCharacter = installPath.chars().anyMatch(Character::isISOControl); + if ("/".equals(installPath) || !installPath.startsWith("/") || installPath.contains("\\") + || containsRelativeSegment || containsControlCharacter) { + throw new IllegalArgumentException(productName + " 安装目录必须是非根目录的规范 Linux 绝对路径"); + } + } + private void migrateDatabase(MigrationRequest request, MigrationJob job) throws Exception { - job.log("开始通过内置 MySQL JDBC 工具迁移数据库。"); + job.log("开始迁移数据库。"); databaseMigrator.migrate(request.sourceDatabase(), request.targetDatabase(), job); job.log("数据库迁移完成。"); } private void cleanup(ServerInfo server, String remoteFiles, MigrationJob job) { + // 本地归档由 deleteQuietly 负责；只有远程分支会在服务器 /tmp 中创建同名文件。 + if (isLocalHost(server.host())) { + return; + } try { ssh.execute(server, "rm -f " + ShellEscaper.quote(remoteFiles), job); } catch (Exception e) { @@ -120,6 +426,50 @@ private String safeMessage(Exception exception) { return message == null || message.isBlank() ? exception.getClass().getSimpleName() : message; } + /** + * 将包装异常和底层异常按发生顺序展示，避免页面只能看到“脚本执行失败”等上层通用信息。 + * 使用有序集合去重，并限制异常链深度，防止驱动产生循环 cause 时日志无限增长。 + */ + private String detailedMessage(Throwable exception) { + Set reasons = new LinkedHashSet<>(); + Throwable current = exception; + int depth = 0; + while (current != null && depth++ < 10) { + String message = current.getMessage(); + String detail = message == null || message.isBlank() + ? current.getClass().getSimpleName() + : current.getClass().getSimpleName() + "：" + message; + reasons.add(detail); + current = current.getCause(); + } + return String.join("；根因：", reasons); + } + + private void logElapsed(MigrationJob job, String operation, long startedNanos) { + job.log(operation + "耗时：" + formatElapsed(startedNanos) + "。"); + } + + private String formatElapsed(long startedNanos) { + long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startedNanos); + if (elapsedMillis < 1_000) { + return elapsedMillis + " 毫秒"; + } + return String.format(Locale.ROOT, "%.2f 秒", elapsedMillis / 1_000.0); + } + + private String formatBytes(long bytes) { + if (bytes < 1_024) { + return bytes + " B"; + } + if (bytes < 1_024L * 1_024L) { + return String.format(Locale.ROOT, "%.2f KiB", bytes / 1_024.0); + } + if (bytes < 1_024L * 1_024L * 1_024L) { + return String.format(Locale.ROOT, "%.2f MiB", bytes / (1_024.0 * 1_024.0)); + } + return String.format(Locale.ROOT, "%.2f GiB", bytes / (1_024.0 * 1_024.0 * 1_024.0)); + } + private void deleteQuietly(Path path) { if (path == null) { return; @@ -127,7 +477,7 @@ private void deleteQuietly(Path path) { try { Files.deleteIfExists(path); } catch (Exception ignored) { - // The operating system will eventually clean the temporary directory. + // 任务结果不应被临时文件清理失败覆盖；操作系统会继续按策略清理临时目录。 } } } diff --git a/src/main/java/com/dataease/migration/service/PluginMigrationService.java b/src/main/java/com/dataease/migration/service/PluginMigrationService.java index 0999903..cdb5ce6 100644 --- a/src/main/java/com/dataease/migration/service/PluginMigrationService.java +++ b/src/main/java/com/dataease/migration/service/PluginMigrationService.java @@ -3,6 +3,7 @@ import com.dataease.migration.model.DatabaseInfo; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.databind.node.ObjectNode; import org.springframework.stereotype.Service; import java.io.IOException; @@ -16,20 +17,40 @@ import java.util.ArrayList; import java.util.List; import java.util.Map; +import java.util.Set; +import java.util.stream.Stream; import java.util.zip.ZipFile; +/** + * 将迁移工具随包发布的 V3 插件复制信息写入目标库。 + * + *

迁移时同时使用 module_name 和 name 识别同一插件，兼容 V2 旧元数据与 V3 稳定模块名。 + * 数据库中的 driverPath 会重写到本次目标安装目录，保证迁移结果可部署在非 /opt/dataease3.0 路径。

+ */ @Service public class PluginMigrationService { - private static final Map PLUGIN_IDS = Map.of( - "飞书多维表格插件", new PluginId(1274106050030735360L, 1783937019703L), - "Apache Hive数据源插件", new PluginId(1274106098604969984L, 1783937031285L), - "达梦数据源插件", new PluginId(1274106127038156800L, 1783937038064L) + // 通用数据源插件在基础升级完成后写入；driverDirectory 为插件解压 JDBC 驱动的相对目录。 + private static final Map BASE_PLUGIN_IDS = Map.of( + "飞书多维表格插件", new PluginId(1274106050030735360L, 1783937019703L, null), + "Apache Hive数据源插件", new PluginId(1274106098604969984L, 1783937031285L, "hiveDriver"), + "达梦数据源插件", new PluginId(1274106127038156800L, 1783937038064L, "dmDriver") ); - private static final List PLUGIN_JARS = List.of( + private static final Map SYNC_PLUGIN_IDS = Map.of( + "PostgreSQL 目标数据源插件", new PluginId(1284970020220309504L, 1786527192040L, "sync"), + "PostgreSQL 源数据源插件", new PluginId(1284970035768594432L, 1786697211956L, "sync") + ); + private static final Set SUPPORTED_PLUGIN_NAMES = Stream.concat( + BASE_PLUGIN_IDS.keySet().stream(), SYNC_PLUGIN_IDS.keySet().stream() + ).collect(java.util.stream.Collectors.toUnmodifiableSet()); + private static final List BASE_PLUGIN_JARS = List.of( "lark-backend-3.0.0.jar", "hive-backend-3.0.0.jar", "dm-backend-3.0.0.jar" ); + private static final List SYNC_PLUGIN_JARS = List.of( + "postgresql-backend-source-3.0.0.jar", + "postgresql-backend-sink-3.0.0.jar" + ); private final ObjectMapper objectMapper; @@ -38,32 +59,79 @@ public PluginMigrationService(ObjectMapper objectMapper) { } public List pluginJars() { - return PLUGIN_JARS.stream().map(jar -> Path.of("plugins", jar)).toList(); + // 文件迁移阶段必须同时上传通用插件和同步管理插件，否则数据库元数据存在但目标端没有插件 JAR。 + return Stream.concat(basePluginJars().stream(), syncPluginJars().stream()).toList(); + } + + List basePluginJars() { + return BASE_PLUGIN_JARS.stream().map(jar -> Path.of("plugins", jar)).toList(); } - public void updatePlugins(DatabaseInfo target, MigrationJob job) throws Exception { - List plugins = loadPlugins(); + List syncPluginJars() { + return SYNC_PLUGIN_JARS.stream().map(jar -> Path.of("plugins", jar)).toList(); + } + + public void updatePlugins(DatabaseInfo target, String targetInstallPath, MigrationJob job) throws Exception { + updatePlugins(target, targetInstallPath, job, BASE_PLUGIN_IDS, basePluginJars(), true); + job.log("通用插件数据更新完成。"); + } + + public void updateSyncPlugins(DatabaseInfo target, String targetInstallPath, MigrationJob job) throws Exception { + updatePlugins(target, targetInstallPath, job, SYNC_PLUGIN_IDS, syncPluginJars(), false); + job.log("同步管理插件数据更新完成。"); + } + + private void updatePlugins(DatabaseInfo target, String targetInstallPath, MigrationJob job, + Map pluginIds, List pluginJars, + boolean reportUnsupportedPlugins) throws Exception { + List plugins = loadPlugins(pluginIds, pluginJars, targetInstallPath); + List unsupportedPlugins; try (Connection connection = DriverManager.getConnection(target.jdbcUrl(), target.username(), target.password())) { - List unsupportedPlugins = findUnsupportedPlugins(connection); - updateSupportedPlugins(connection, plugins); - for (String name : unsupportedPlugins) { - job.log("请升级插件：" + name); + boolean originalAutoCommit = connection.getAutoCommit(); + connection.setAutoCommit(false); + try { + updateSupportedPlugins(connection, plugins); + // 更新后再检查，避免仅展示名称发生变化的已支持插件被误报为“不支持”。 + unsupportedPlugins = reportUnsupportedPlugins + ? findUnsupportedPlugins(connection) : List.of(); + connection.commit(); + } catch (SQLException | RuntimeException e) { + try { + connection.rollback(); + } catch (SQLException rollbackException) { + e.addSuppressed(rollbackException); + } + throw e; + } finally { + connection.setAutoCommit(originalAutoCommit); } } - job.log("插件数据更新完成。"); + for (String name : unsupportedPlugins) { + job.log("请升级插件：" + name); + } } - private List loadPlugins() throws IOException { + private List loadPlugins(Map pluginIds, List pluginJars, + String targetInstallPath) throws IOException { List plugins = new ArrayList<>(); - for (Path jar : pluginJars()) { + for (Path jar : pluginJars) { try (ZipFile zip = new ZipFile(jar.toFile())) { String metadata = readEntry(zip, "plugin/", ".json"); JsonNode root = objectMapper.readTree(metadata); String name = requiredText(root, "name"); - PluginId pluginId = PLUGIN_IDS.get(name); + PluginId pluginId = pluginIds.get(name); if (pluginId == null) { throw new IllegalStateException("未配置插件数据：" + name); } + JsonNode config = root.required("config").deepCopy(); + if (!config.isObject()) { + throw new IllegalStateException("插件配置格式无效：" + name); + } + if (pluginId.driverDirectory() != null) { + // 插件包内的默认路径通常指向 /opt/dataease3.0；迁移目标允许自定义安装目录，不能沿用默认值。 + ((ObjectNode) config).put("driverPath", + resolveDriverPath(targetInstallPath, pluginId.driverDirectory())); + } plugins.add(new PluginDefinition( pluginId.id(), name, @@ -72,7 +140,7 @@ private List loadPlugins() throws IOException { pluginId.installTime(), requiredText(root, "flag"), requiredText(root, "developer"), - objectMapper.writeValueAsString(root.required("config")), + objectMapper.writeValueAsString(config), requiredText(root, "requireVersion"), requiredText(root, "moduleName"), jar.getFileName().toString() @@ -82,13 +150,23 @@ private List loadPlugins() throws IOException { return plugins; } + /** + * 生成写入 xpack_plugin.config 的绝对驱动目录，与 V3 插件运行时的 drivers/plugin 目录约定保持一致。 + */ + static String resolveDriverPath(String targetInstallPath, String driverDirectory) { + // targetInstallPath 是目标 Linux 服务器路径，不能使用迁移工具所在操作系统的 Path 规则拼接； + // 否则 Windows 上运行迁移工具时可能把反斜杠写入 Linux 目标库。 + String normalizedInstallPath = targetInstallPath.replaceFirst("/+$", ""); + return normalizedInstallPath + "/drivers/plugin/" + driverDirectory; + } + private List findUnsupportedPlugins(Connection connection) throws SQLException { List names = new ArrayList<>(); try (PreparedStatement statement = connection.prepareStatement("SELECT name FROM xpack_plugin"); ResultSet result = statement.executeQuery()) { while (result.next()) { String name = result.getString(1); - if (!PLUGIN_IDS.containsKey(name)) { + if (!SUPPORTED_PLUGIN_NAMES.contains(name)) { names.add(name); } } @@ -97,28 +175,74 @@ private List findUnsupportedPlugins(Connection connection) throws SQLExc } private void updateSupportedPlugins(Connection connection, List plugins) throws SQLException { - String sql = "UPDATE xpack_plugin SET id = ?, icon = ?, version = ?, install_time = ?, flag = ?, " - + "developer = ?, config = ?, require_version = ?, module_name = ?, jar_name = ? WHERE name = ?"; - try (PreparedStatement statement = connection.prepareStatement(sql)) { + // module_name 是 V2/V3 运行时使用的稳定业务键；name 则兼容旧迁移逻辑和历史展示名称。 + // 两者若命中不同记录，说明目标库已有冲突数据。此时中止比静默插入重复插件更安全。 + String findSql = "SELECT id FROM xpack_plugin WHERE module_name = ? OR name = ?"; + String updateSql = "UPDATE xpack_plugin SET name = ?, icon = ?, version = ?, install_time = ?, flag = ?, " + + "developer = ?, config = ?, require_version = ?, module_name = ?, jar_name = ? WHERE id = ?"; + String insertSql = "INSERT INTO xpack_plugin (id, name, icon, version, install_time, flag, developer, config, " + + "require_version, module_name, jar_name) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"; + try (PreparedStatement find = connection.prepareStatement(findSql); + PreparedStatement update = connection.prepareStatement(updateSql); + PreparedStatement insert = connection.prepareStatement(insertSql)) { for (PluginDefinition plugin : plugins) { - statement.setLong(1, plugin.id()); - statement.setString(2, plugin.icon()); - statement.setString(3, plugin.version()); - statement.setLong(4, plugin.installTime()); - statement.setString(5, plugin.flag()); - statement.setString(6, plugin.developer()); - statement.setString(7, plugin.config()); - statement.setString(8, plugin.requireVersion()); - statement.setString(9, plugin.moduleName()); - statement.setString(10, plugin.jarName()); - statement.setString(11, plugin.name()); - if (statement.executeUpdate() != 1) { - throw new SQLException("未找到待更新的插件：" + plugin.name()); + Long existingId = findExistingPluginId(find, plugin); + if (existingId != null) { + // 保留 V2 记录主键，只更新插件元数据，避免无必要地改变已有记录身份。 + bindUpdate(update, plugin, existingId); + update.executeUpdate(); + } else { + bindInsert(insert, plugin); + insert.executeUpdate(); } } } } + private Long findExistingPluginId(PreparedStatement statement, PluginDefinition plugin) throws SQLException { + statement.setString(1, plugin.moduleName()); + statement.setString(2, plugin.name()); + try (ResultSet result = statement.executeQuery()) { + Long id = null; + while (result.next()) { + if (id != null) { + throw new SQLException("插件名称或模块存在冲突数据：" + plugin.name() + + "（" + plugin.moduleName() + "）"); + } + id = result.getLong(1); + } + return id; + } + } + + private void bindUpdate(PreparedStatement statement, PluginDefinition plugin, long existingId) throws SQLException { + statement.setString(1, plugin.name()); + statement.setString(2, plugin.icon()); + statement.setString(3, plugin.version()); + statement.setLong(4, plugin.installTime()); + statement.setString(5, plugin.flag()); + statement.setString(6, plugin.developer()); + statement.setString(7, plugin.config()); + statement.setString(8, plugin.requireVersion()); + statement.setString(9, plugin.moduleName()); + statement.setString(10, plugin.jarName()); + statement.setLong(11, existingId); + } + + private void bindInsert(PreparedStatement statement, PluginDefinition plugin) throws SQLException { + statement.setLong(1, plugin.id()); + statement.setString(2, plugin.name()); + statement.setString(3, plugin.icon()); + statement.setString(4, plugin.version()); + statement.setLong(5, plugin.installTime()); + statement.setString(6, plugin.flag()); + statement.setString(7, plugin.developer()); + statement.setString(8, plugin.config()); + statement.setString(9, plugin.requireVersion()); + statement.setString(10, plugin.moduleName()); + statement.setString(11, plugin.jarName()); + } + private String readEntry(ZipFile zip, String directory, String suffix) throws IOException { var entries = zip.entries(); while (entries.hasMoreElements()) { @@ -140,7 +264,7 @@ private String requiredText(JsonNode root, String field) { return value.textValue(); } - private record PluginId(long id, long installTime) { + private record PluginId(long id, long installTime, String driverDirectory) { } private record PluginDefinition(long id, String name, String icon, String version, long installTime, String flag, diff --git a/src/main/java/com/dataease/migration/service/SyncManagementMigrationService.java b/src/main/java/com/dataease/migration/service/SyncManagementMigrationService.java new file mode 100644 index 0000000..cd0bb40 --- /dev/null +++ b/src/main/java/com/dataease/migration/service/SyncManagementMigrationService.java @@ -0,0 +1,382 @@ +package com.dataease.migration.service; + +import com.dataease.migration.model.DatabaseInfo; +import org.springframework.core.io.ClassPathResource; +import org.springframework.core.io.support.EncodedResource; +import org.springframework.jdbc.datasource.init.ScriptUtils; +import org.springframework.stereotype.Service; + +import java.nio.charset.StandardCharsets; +import java.sql.Connection; +import java.sql.DriverManager; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.sql.SQLException; +import java.sql.Statement; +import java.util.ArrayList; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.TreeSet; +import java.util.concurrent.TimeUnit; + +/** + * V2 到 V3 的同步管理专项迁移入口。 + * + *

通用 upgrade.sql 只负责公共表结构。本服务单独识别同步模块是否存在，补充 JPA 实体需要的 + * datasource_role，转换任务 JSON 和运行状态，并安装 PostgreSQL 源/目标插件。专项逻辑与通用升级隔离， + * 可避免没有同步管理数据的环境受到影响。

+ */ +@Service +public class SyncManagementMigrationService { + private static final String SYNC_UPGRADE_SCRIPT = "sync-upgrade.sql"; + /** + * V3 会把 parameter 反序列化为包含 source.datasource 和 target.datasource 的任务对象。 + * 使用 CASE 保证非法 JSON 不会继续进入 JSON_EXTRACT，兼容历史 longtext/json 两种列类型。 + */ + private static final String INVALID_TASK_PARAMETER_QUERY = """ + SELECT id, + `_name`, + CASE + WHEN parameter IS NULL THEN 'parameter 为空' + WHEN JSON_VALID(parameter) = 0 THEN 'parameter 不是合法 JSON' + WHEN COALESCE(JSON_TYPE(JSON_EXTRACT(parameter, '$.source.datasource')), '') <> 'OBJECT' + AND COALESCE(JSON_TYPE(JSON_EXTRACT(parameter, '$.target.datasource')), '') <> 'OBJECT' + THEN '缺少 source.datasource 和 target.datasource 对象' + WHEN COALESCE(JSON_TYPE(JSON_EXTRACT(parameter, '$.source.datasource')), '') <> 'OBJECT' + THEN '缺少 source.datasource 对象' + WHEN COALESCE(JSON_TYPE(JSON_EXTRACT(parameter, '$.target.datasource')), '') <> 'OBJECT' + THEN '缺少 target.datasource 对象' + ELSE '未知参数结构异常' + END AS invalid_reason + FROM per_sync_task_info + WHERE CASE + WHEN parameter IS NULL OR JSON_VALID(parameter) = 0 THEN 1 + WHEN COALESCE(JSON_TYPE(JSON_EXTRACT(parameter, '$.source.datasource')), '') <> 'OBJECT' THEN 1 + WHEN COALESCE(JSON_TYPE(JSON_EXTRACT(parameter, '$.target.datasource')), '') <> 'OBJECT' THEN 1 + ELSE 0 + END = 1 + ORDER BY id + """; + // 四张表共同构成可迁移的 V2 同步管理数据；只存在部分表通常表示源库不完整，不能静默继续。 + static final Set REQUIRED_TABLES = Set.of( + "per_sync_datasource", + "per_sync_task_info", + "per_sync_task_log", + "per_sync_task_lock" + ); + + private final PluginMigrationService pluginMigrationService; + + public SyncManagementMigrationService(PluginMigrationService pluginMigrationService) { + this.pluginMigrationService = pluginMigrationService; + } + + public void execute(DatabaseInfo target, String targetInstallPath, MigrationJob job) throws Exception { + job.log("开始同步管理专项迁移。"); + try (Connection connection = DriverManager.getConnection(target.jdbcUrl(), target.username(), target.password())) { + Set existingTables = findExistingTables(connection); + SyncSchemaStatus schemaStatus = classifyTables(existingTables); + if (schemaStatus == SyncSchemaStatus.ABSENT) { + job.log("目标数据库不存在 V2 同步管理表，跳过同步数据转换。"); + } else if (schemaStatus == SyncSchemaStatus.PARTIAL) { + // 部分表缺失时继续运行会产生无法被 JPA 正确读取的半迁移数据，必须中止并明确列出缺表。 + Set missingTables = new TreeSet<>(REQUIRED_TABLES); + missingTables.removeAll(existingTables); + throw new SQLException("V2 同步管理表不完整，缺少：" + String.join("、", missingTables)); + } else { + migrateSyncData(connection, job); + } + } + // 即使源库没有同步业务表，也要安装同步插件，使全新的 V3 环境能够创建 PostgreSQL 源/目标数据源。 + pluginMigrationService.updateSyncPlugins(target, targetInstallPath, job); + job.log("同步管理专项迁移完成。"); + } + + static SyncSchemaStatus classifyTables(Set existingTables) { + long matches = REQUIRED_TABLES.stream().filter(existingTables::contains).count(); + if (matches == 0) { + return SyncSchemaStatus.ABSENT; + } + return matches == REQUIRED_TABLES.size() ? SyncSchemaStatus.READY : SyncSchemaStatus.PARTIAL; + } + + /** + * 迁移同步管理数据时，把不可回滚的 DDL 与可回滚的 DML 分开：先校验并补齐字段，再在一个事务中 + * 转换任务、数据源、日志和锁，最后收紧字段约束。各阶段记录耗时，便于定位大数据量环境的瓶颈。 + */ + private void migrateSyncData(Connection connection, MigrationJob job) throws SQLException { + long stepStartedNanos = System.nanoTime(); + validateTaskParameters(connection, job); + logElapsed(job, "同步任务参数校验", stepStartedNanos); + + stepStartedNanos = System.nanoTime(); + SyncMigrationStats before = readStats(connection, false); + logElapsed(job, "同步管理迁移前统计", stepStartedNanos); + // MySQL DDL 会隐式提交，因此先以幂等方式补列，再在独立事务中执行所有数据转换。 + stepStartedNanos = System.nanoTime(); + ensureDatasourceRoleColumn(connection, job); + logElapsed(job, "同步数据源角色字段准备", stepStartedNanos); + + boolean originalAutoCommit = connection.getAutoCommit(); + connection.setAutoCommit(false); + stepStartedNanos = System.nanoTime(); + int migratedRunningLogCount; + try { + ScriptUtils.executeSqlScript(connection, + new EncodedResource(new ClassPathResource(SYNC_UPGRADE_SCRIPT), StandardCharsets.UTF_8)); + migratedRunningLogCount = markRunningLogsConnectionLost(connection); + connection.commit(); + } catch (SQLException | RuntimeException e) { + connection.rollback(); + throw e; + } finally { + connection.setAutoCommit(originalAutoCommit); + } + logElapsed(job, "同步管理数据转换", stepStartedNanos); + + // 转换完成后移除迁移期默认值 0，后续 V3/JPA 写入必须明确指定源端(1)或目标端(2)。 + stepStartedNanos = System.nanoTime(); + enforceDatasourceRoleColumn(connection); + logElapsed(job, "同步数据源角色字段约束", stepStartedNanos); + + stepStartedNanos = System.nanoTime(); + SyncMigrationStats after = readStats(connection, true); + logElapsed(job, "同步管理迁移后校验", stepStartedNanos); + job.log("同步管理迁移前：数据源 " + before.datasourceCount() + "，PG 类型 " + before.postgresqlCount() + + "，任务 " + before.taskCount() + "，运行中任务 " + before.runningTaskCount() + + "，有效锁 " + before.activeLockCount() + "。"); + job.log("同步管理迁移后：数据源 " + after.datasourceCount() + "，PostgreSQL 类型 " + + after.postgresqlCount() + "，源端 " + after.sourceCount() + "，目标端 " + after.targetCount() + + "，异常角色 " + after.invalidRoleCount() + "，任务 " + after.taskCount() + + "，运行中任务 " + after.runningTaskCount() + "，已终止运行中日志 " + migratedRunningLogCount + + "，有效锁 " + after.activeLockCount() + "。"); + } + + /** + * 历史日志可能达到百万级。这里让 UPDATE 自身完成唯一一次必要扫描，并直接使用受影响行数， + * 不再为迁移前/后日志统计额外执行 COUNT 全表扫描。列举 V2 常见大小写而不对列执行 UPPER， + * 使已有 status 索引仍有机会参与更新定位。 + */ + private int markRunningLogsConnectionLost(Connection connection) throws SQLException { + try (PreparedStatement statement = connection.prepareStatement(""" + UPDATE per_sync_task_log + SET status = 'CONNECTION_LOST', + executor_end_time = COALESCE( + executor_end_time, + FLOOR(UNIX_TIMESTAMP(CURRENT_TIMESTAMP(3)) * 1000)) + WHERE status IN ('RUNNING', 'Running', 'running') + """)) { + return statement.executeUpdate(); + } + } + + /** + * 在任何同步表 DDL/DML 之前校验任务参数。异常任务不能静默跳过，否则迁移任务会显示成功， + * 但 V3 在任务列表反序列化 parameter 时仍会失败。 + */ + private void validateTaskParameters(Connection connection, MigrationJob job) throws SQLException { + List invalidTasks = findInvalidSyncTasks(connection); + if (!invalidTasks.isEmpty()) { + job.log("同步任务参数校验失败，以下任务无法被 V3/JPA 正确读取："); + for (InvalidSyncTask task : invalidTasks) { + // 不输出 parameter 原文，避免把数据库密码等连接信息写入迁移日志。 + job.log("异常同步任务：ID=" + sanitizeLogValue(task.id()) + + "，名称=" + sanitizeLogValue(task.name()) + + "，原因=" + task.reason()); + } + throw new SQLException("发现 " + invalidTasks.size() + + " 个异常同步任务；任务 ID、名称和具体原因已在上方日志列出。" + + "请先在 V2 中修复或删除这些任务，再使用全新目标库重新迁移"); + } + job.log("同步任务参数 JSON 及源/目标数据源结构校验通过。"); + } + + private List findInvalidSyncTasks(Connection connection) throws SQLException { + List invalidTasks = new ArrayList<>(); + try (Statement statement = connection.createStatement(); + ResultSet result = statement.executeQuery(INVALID_TASK_PARAMETER_QUERY)) { + while (result.next()) { + invalidTasks.add(new InvalidSyncTask( + result.getString("id"), + result.getString("_name"), + result.getString("invalid_reason") + )); + } + } + return invalidTasks; + } + + /** + * 任务名称属于用户输入，去掉换行和控制字符，避免一条异常任务伪装成多条迁移日志。 + */ + private String sanitizeLogValue(String value) { + if (value == null) { + return "(空)"; + } + return value.replaceAll("\\p{Cntrl}", "?"); + } + + private Set findExistingTables(Connection connection) throws SQLException { + String sql = """ + SELECT table_name + FROM information_schema.tables + WHERE table_schema = DATABASE() + AND table_name IN (?, ?, ?, ?) + """; + Set tables = new HashSet<>(); + try (PreparedStatement statement = connection.prepareStatement(sql)) { + int index = 1; + for (String table : REQUIRED_TABLES) { + statement.setString(index++, table); + } + try (ResultSet result = statement.executeQuery()) { + while (result.next()) { + tables.add(result.getString(1)); + } + } + } + return tables; + } + + private void ensureDatasourceRoleColumn(Connection connection, MigrationJob job) throws SQLException { + String sql = """ + SELECT COUNT(*) + FROM information_schema.columns + WHERE table_schema = DATABASE() + AND table_name = 'per_sync_datasource' + AND column_name = 'datasource_role' + """; + if (queryCount(connection, sql) > 0) { + job.log("同步数据源角色字段已存在，继续执行幂等数据转换。"); + return; + } + try (Statement statement = connection.createStatement()) { + statement.execute(""" + ALTER TABLE per_sync_datasource + ADD COLUMN datasource_role INT NOT NULL DEFAULT 0 COMMENT '1:源数据库 2:目标数据库' + """); + } + job.log("已添加同步数据源角色字段 datasource_role。"); + } + + /** + * 只在约束确实不满足时执行 DDL。首次迁移添加的列已经是 NOT NULL，只需删除临时默认值； + * 直接再次 MODIFY COLUMN 可能重建整张数据源表，幂等重跑时没有必要承担这项开销。 + */ + private void enforceDatasourceRoleColumn(Connection connection) throws SQLException { + String columnSql = """ + SELECT is_nullable, column_default + FROM information_schema.columns + WHERE table_schema = DATABASE() + AND table_name = 'per_sync_datasource' + AND column_name = 'datasource_role' + """; + boolean nullable; + boolean hasDefault; + try (Statement query = connection.createStatement(); + ResultSet result = query.executeQuery(columnSql)) { + if (!result.next()) { + throw new SQLException("同步数据源角色字段 datasource_role 不存在"); + } + nullable = "YES".equalsIgnoreCase(result.getString("is_nullable")); + hasDefault = result.getObject("column_default") != null; + } + if (nullable) { + try (Statement alter = connection.createStatement()) { + alter.execute(""" + ALTER TABLE per_sync_datasource + MODIFY COLUMN datasource_role INT NOT NULL COMMENT '1:源数据库 2:目标数据库' + """); + } + } else if (hasDefault) { + // 只移除迁移期默认值时使用元数据级 DDL，避免 MODIFY COLUMN 触发表重建。 + try (Statement alter = connection.createStatement()) { + alter.execute(""" + ALTER TABLE per_sync_datasource + ALTER COLUMN datasource_role DROP DEFAULT + """); + } + } + } + + /** + * 按表聚合后 CROSS JOIN 成一行，减少 JDBC 往返，并确保每张业务表每轮最多扫描一次。 + * includeRoles=false 用于补列前的 V2 统计，此时 SQL 不能引用尚不存在的 datasource_role。 + */ + private SyncMigrationStats readStats(Connection connection, boolean includeRoles) throws SQLException { + String roleStats = includeRoles + ? """ + COALESCE(SUM(datasource_role = 1), 0) AS source_count, + COALESCE(SUM(datasource_role = 2), 0) AS target_count, + COALESCE(SUM(datasource_role NOT IN (1, 2)), 0) AS invalid_role_count + """ + : """ + 0 AS source_count, + 0 AS target_count, + 0 AS invalid_role_count + """; + String postgresqlType = includeRoles ? "postgresql" : "pg"; + // 百万级日志表不参与统计；其余表每轮只扫描一次，并合并成一次 JDBC 往返。 + String sql = """ + SELECT datasource_stats.*, task_stats.*, lock_stats.* + FROM ( + SELECT COUNT(*) AS datasource_count, + COALESCE(SUM(LOWER(type) = '%s'), 0) AS postgresql_count, + %s + FROM per_sync_datasource + ) datasource_stats + CROSS JOIN ( + SELECT COUNT(*) AS task_count, + COALESCE(SUM(`_status` IN ('RUNNING', 'Running', 'running')), 0) + AS running_task_count + FROM per_sync_task_info + ) task_stats + CROSS JOIN ( + SELECT COALESCE(SUM(expiration_time > CURRENT_TIMESTAMP), 0) AS active_lock_count + FROM per_sync_task_lock + ) lock_stats + """.formatted(postgresqlType, roleStats); + try (Statement statement = connection.createStatement(); + ResultSet result = statement.executeQuery(sql)) { + result.next(); + return new SyncMigrationStats( + result.getInt("datasource_count"), + result.getInt("postgresql_count"), + result.getInt("source_count"), + result.getInt("target_count"), + result.getInt("invalid_role_count"), + result.getInt("task_count"), + result.getInt("running_task_count"), + result.getInt("active_lock_count") + ); + } + } + + private int queryCount(Connection connection, String sql) throws SQLException { + try (Statement statement = connection.createStatement(); + ResultSet result = statement.executeQuery(sql)) { + result.next(); + return result.getInt(1); + } + } + + private void logElapsed(MigrationJob job, String operation, long startedNanos) { + long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startedNanos); + job.log(operation + "耗时：" + elapsedMillis + " 毫秒。"); + } + + enum SyncSchemaStatus { + ABSENT, + READY, + PARTIAL + } + + private record SyncMigrationStats(int datasourceCount, int postgresqlCount, int sourceCount, int targetCount, + int invalidRoleCount, int taskCount, int runningTaskCount, + int activeLockCount) { + } + + private record InvalidSyncTask(String id, String name, String reason) { + } +} diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml index 18199c1..5243853 100644 --- a/src/main/resources/application.yml +++ b/src/main/resources/application.yml @@ -9,6 +9,9 @@ spring: request-timeout: 0 migration: + files: + # 同步物理日志可能很大，必须由实施人员根据磁盘和传输空间显式开启。 + copy-sync-task-logs: ${MIGRATION_COPY_SYNC_TASK_LOGS:false} ssh: connect-timeout-millis: 30000 command-timeout-millis: 3600000 diff --git a/src/main/resources/static/app.css b/src/main/resources/static/app.css index 76dfe35..fb10e0b 100644 --- a/src/main/resources/static/app.css +++ b/src/main/resources/static/app.css @@ -8,6 +8,8 @@ section { background: white; border: 1px solid #dce3ef; border-radius: 8px; marg .grid { display: grid; gap: 16px; grid-template-columns: repeat(2, minmax(0, 1fr)); } label { display: grid; font-size: .9rem; font-weight: 600; gap: 7px; } input { border: 1px solid #aebbd0; border-radius: 4px; box-sizing: border-box; font: inherit; font-weight: 400; padding: 9px; width: 100%; } +.connection-mode p { color: #52627a; font-size: .9rem; margin: 0; } +.connection-mode p + p { margin-top: 10px; } .wide { grid-column: span 2; } button { background: #0969da; border: 0; border-radius: 5px; color: white; cursor: pointer; font: inherit; margin-top: 20px; padding: 10px 20px; } button:disabled { background: #8194ae; cursor: wait; } diff --git a/src/main/resources/static/index.html b/src/main/resources/static/index.html index 3be7b88..1c3ab72 100644 --- a/src/main/resources/static/index.html +++ b/src/main/resources/static/index.html @@ -11,13 +11,18 @@

DataEase 2.0 → 3.0 数据迁移

执行后会删除并重建 DataEase 3.0 的目标数据库。请先确认目标库可被覆盖。

+
+

服务器连接方式

+

工具会分别判断源端和目标端：localhost、127.x、::1 或本机网卡地址直接操作本地安装目录，远程地址通过 SSH 操作。无论使用哪种连接方式，文件、数据库、升级脚本、插件及同步管理都会执行相同的完整迁移流程。

+

同步任务物理日志是否复制由启动参数 --migration.files.copy-sync-task-logs=true 控制，默认关闭；任务日志会显示本次是否启用。

+

DataEase 2.0（源端）

- - - + + + @@ -28,13 +33,13 @@

DataEase 2.0（源端）

DataEase 3.0（目标端）

- - - + + + - +
diff --git a/src/main/resources/sync-upgrade.sql b/src/main/resources/sync-upgrade.sql new file mode 100644 index 0000000..f7724fa --- /dev/null +++ b/src/main/resources/sync-upgrade.sql @@ -0,0 +1,48 @@ +SET SQL_SAFE_UPDATES = 0; + +-- datasource_role 是 V3 JPA 查询源端/目标端列表的判定字段；迁移服务会在执行本脚本前幂等补列。 +-- V2 内置 PostgreSQL 类型为 pg，V3 同步插件统一使用 postgresql。 +UPDATE per_sync_datasource +SET type = 'postgresql' +WHERE LOWER(type) = 'pg'; + +-- V2 只有 Doris 作为内置目标端，其余内置同步数据源均为源端。 +UPDATE per_sync_datasource +SET datasource_role = CASE WHEN LOWER(type) = 'doris' THEN 2 ELSE 1 END; + +-- V2 任务参数保存了完整的数据源快照，需要同步迁移外层类型、快照类型和数据源角色。 +-- 参数已经由迁移服务校验，因此一次完成全部转换，避免对 JSON 大字段连续六次全表扫描和重写。 +-- JSON_REPLACE 不会创建缺失的可选 type 路径；已有 type 按同步任务模型约定均为字符串。 +UPDATE per_sync_task_info +SET parameter = JSON_REPLACE( + JSON_SET(parameter, + '$.source.datasource.datasourceRole', 1, + '$.target.datasource.datasourceRole', 2), + '$.source.type', + IF(LOWER(JSON_UNQUOTE(JSON_EXTRACT(parameter, '$.source.type'))) = 'pg', + 'postgresql', JSON_UNQUOTE(JSON_EXTRACT(parameter, '$.source.type'))), + '$.source.datasource.type', + IF(LOWER(JSON_UNQUOTE(JSON_EXTRACT(parameter, '$.source.datasource.type'))) = 'pg', + 'postgresql', JSON_UNQUOTE(JSON_EXTRACT(parameter, '$.source.datasource.type'))), + '$.target.type', + IF(LOWER(JSON_UNQUOTE(JSON_EXTRACT(parameter, '$.target.type'))) = 'pg', + 'postgresql', JSON_UNQUOTE(JSON_EXTRACT(parameter, '$.target.type'))), + '$.target.datasource.type', + IF(LOWER(JSON_UNQUOTE(JSON_EXTRACT(parameter, '$.target.datasource.type'))) = 'pg', + 'postgresql', JSON_UNQUOTE(JSON_EXTRACT(parameter, '$.target.datasource.type')))) +WHERE JSON_VALID(parameter) + AND JSON_TYPE(JSON_EXTRACT(parameter, '$.source.datasource')) = 'OBJECT' + AND JSON_TYPE(JSON_EXTRACT(parameter, '$.target.datasource')) = 'OBJECT'; + +-- 全新迁移后先暂停定时任务；V2 已停止后遗留的手动运行态则改为完成。 +-- 合并为一次更新，减少任务表扫描和事务日志写入。 +UPDATE per_sync_task_info +SET `_status` = CASE WHEN scheduler_type <> 'NONE' THEN 'SUSPEND' ELSE 'DONE' END, + trigger_next_time = -1 +WHERE scheduler_type <> 'NONE' + OR (scheduler_type = 'NONE' AND `_status` IN ('RUNNING', 'Running', 'running')); + +-- 历史日志量可能达到百万级。运行中日志转换由迁移服务单独执行并读取受影响行数， +-- 避免为了迁移前后统计再对 per_sync_task_log 做额外全表扫描。 +UPDATE per_sync_task_lock +SET expiration_time = DATE_SUB(CURRENT_TIMESTAMP, INTERVAL 1 SECOND);