diff --git a/.github/workflows/cross-platform.yml b/.github/workflows/cross-platform.yml index 84c07f2..2fc7c31 100644 --- a/.github/workflows/cross-platform.yml +++ b/.github/workflows/cross-platform.yml @@ -11,6 +11,8 @@ on: - "tests/Run-CrossPlatformLiveTests.sh" - "tests/Run-PostgreSqlVersionMigration.py" - "tests/PostgreSqlVersionMigrationHarness/**" + - "tests/Run-MySqlLogicalMigration.py" + - "tests/MySqlLogicalMigrationHarness/**" - "tests/Run-DesktopProfileOperationTests.sh" - "tests/DesktopProfileOperationHarness/**" - "tests/Test-LinuxCrossPlatformPackage.sh" @@ -28,6 +30,8 @@ on: - "tests/Run-CrossPlatformLiveTests.sh" - "tests/Run-PostgreSqlVersionMigration.py" - "tests/PostgreSqlVersionMigrationHarness/**" + - "tests/Run-MySqlLogicalMigration.py" + - "tests/MySqlLogicalMigrationHarness/**" - "tests/Run-DesktopProfileOperationTests.sh" - "tests/DesktopProfileOperationHarness/**" - "tests/Test-LinuxCrossPlatformPackage.sh" @@ -89,6 +93,9 @@ jobs: - name: Build formal PostgreSQL version fixture run: dotnet build tests/PostgreSqlVersionMigrationHarness/Review.csproj -c Release + - name: Build formal MySQL logical migration fixture + run: dotnet build tests/MySqlLogicalMigrationHarness/Review.csproj -c Release + - name: Run isolated Avalonia profile operation handlers if: matrix.name == 'Linux x64' run: ./tests/Run-DesktopProfileOperationTests.sh @@ -97,6 +104,22 @@ jobs: if: matrix.live_tests == true run: ./tests/Run-CrossPlatformLiveTests.sh + - name: Run isolated MySQL 8.4 and MariaDB 11.4 bidirectional logical migration + if: matrix.live_tests == true + run: | + docker pull mysql:8.4 + docker pull mariadb:11.4 + docker pull postgres:16-alpine + python3 tests/Run-MySqlLogicalMigration.py "$RUNNER_TEMP/mysqlpunk-logical-$GITHUB_RUN_ID-$GITHUB_RUN_ATTEMPT" + + - name: Preserve synthetic logical migration evidence + if: always() && matrix.live_tests == true + uses: actions/upload-artifact@v4 + with: + name: mysql-logical-migration-evidence + path: ${{ runner.temp }}/mysqlpunk-logical-${{ github.run_id }}-${{ github.run_attempt }}/ + if-no-files-found: ignore + - name: Build self-contained installable archives run: | package_version=$(sed -n 's/.*AssemblyFileVersion("\([0-9.]*\)").*/\1/p' mySQLPunk/Properties/AssemblyInfo.cs | tail -n 1) diff --git a/CHANGELOG.md b/CHANGELOG.md index 44d07e9..a1a7e55 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,7 @@ ## [Unreleased] ### 🚀 新增功能 +- **MySQL／MariaDB ↔ PostgreSQL 受控邏輯搬移**：加入 signed BIGINT、LONGTEXT NO PAD binary／text C、LONGBLOB／bytea、整數單欄／複合主鍵與具名普通索引。MySQL 8.4／MariaDB 11.4 目標用本輪 staging、分批 DML 交易、schema／SHA256 校驗後單次 RENAME 發布並再次驗證，不假稱 DDL 可完整回滾。失敗／取消列出殘留物件，發布結果不明禁止盲重試，不自動刪除。其他型別、文字前綴鍵、自訂 PK 名稱與相依物件明確拒絕；仍不是任意引擎／版本無損搬移，發版繼續暫停。 - **邏輯搬移的文字、主鍵與索引子集**：SQLite STRICT ↔ PostgreSQL 保留有效 UTF-8 TEXT（BINARY/C）、明確非自增單欄／複合主鍵、普通升冪索引與欄位順序。預覽包含索引 DDL，來源變更失效、取消、缺索引或內容不符仍整批回滾；FK、其他定序、default、identity、特殊索引依舊預檢拒絕，不宣稱所有跨引擎組合完成。 - **可驗證邏輯搬移的第一階段**：SQL Server 專用相容工具加入受限雙向跨版本；Avalonia 新增 SQLite STRICT ↔ PostgreSQL 的整數／二進位搬移，預覽來源／目標版本、schema/data 轉換與阻擋原因，確認後才寫入另建空目標。單一交易、分批、內容校驗與不確定狀態禁止重試，不將一般資料傳輸或原生備份宣稱為無限制跨版本／跨引擎還原。 diff --git a/docs/LOGICAL_MIGRATION_MATRIX.md b/docs/LOGICAL_MIGRATION_MATRIX.md index 79f727e..3885fee 100644 --- a/docs/LOGICAL_MIGRATION_MATRIX.md +++ b/docs/LOGICAL_MIGRATION_MATRIX.md @@ -11,6 +11,30 @@ | SQLite → PostgreSQL | SQLite 3.37–3.53 → PostgreSQL 14–17 | INTEGER → bigint、BLOB → bytea、TEXT BINARY → text C；非自增 PK、普通索引 | 隔離 CI PostgreSQL 16 fixture 加入文字／鍵／索引雙向 round-trip；只以這次新 head 結果為證據 | | PostgreSQL → SQLite | PostgreSQL 14–17 → SQLite 3.37–3.53 | 上述型別反向映射；PK → WITHOUT ROWID、普通索引 | PG17 實際 roundtrip 與既有 PG16 fixture；PG15 未實機跑 | | PostgreSQL → PostgreSQL | 14–17 雙向及同版本 | 同一 bigint/bytea/text C、PK／索引子集與全部預檢 | 真實 17.10 ↔ 14.24、17.10 ↔ 16.15；不是全部版本配對驗證 | +| MySQL／MariaDB → PostgreSQL | MySQL 8.4.x／MariaDB 11.4.x → PG14–17 | signed BIGINT／LONGTEXT NO PAD binary／LONGBLOB，整數 PK／普通索引；PG 目標單交易 | 正式路徑實際 8.4.11 → PG16.15、11.4.13 → PG16.15，各 1007 列＋空表；其他 PG 版本配對未跑 | +| PostgreSQL → MySQL／MariaDB | PG14–17 → MySQL 8.4.x／MariaDB 11.4.x | 上述反向映射，專屬 staging＋校驗＋單一 multi-table RENAME＋發布後校驗 | 正式路徑實際 PG16.15 → 8.4.11／11.4.13；不是全版本、全型別或一般 DDL atomic 保證 | + +## 第三階段：MySQL／MariaDB ↔ PostgreSQL + +這是正式 Core 與 Avalonia 可選取的邏輯搬移，不是只加入引擎白名單。MySQL 的 DDL 隱含提交，不沿用 SQLite／PG 的「schema＋rows 單一交易」說法： + +1. 重新預檢來源 fingerprint 與空目標，確認版本、既有 catalog 可見性、InnoDB Dynamic、`lower_case_table_names=0` 與 utf8mb4 資料庫定序。只接受 MySQL `utf8mb4_0900_bin`／MariaDB `utf8mb4_nopad_bin`，拒絕 PAD SPACE 或其他定序。不修改帳號權限、伺服器設定或既有物件。 +2. MySQL 來源使用 RepeatableRead＋完整有界 `LOCK IN SHARE MODE` 讀取，鎖讀過的資料及 metadata 到搬移結束，最終重讀來源 schema／內容。PG 來源維持既有 SHARE locks＋最後重驗。確認後同列數變動也使 fingerprint 失效。最多 100 表／100,000 列／64 MiB，單欄 1 MiB；先 OCTET_LENGTH／CASE 限額，才讀取完整欄位。 +3. MySQL 目標本輪生成 `_mp_<隨機 GUID>_<序號>` 專屬 staging 表，CREATE DDL 確實已隱含提交。DML 在另一交易中分批寫入，最多 100 列，並依目標 `max_allowed_packet` 與 UTF8／binary escaping 保守 byte 預算縮批；單列不合預算先拒絕。每次 CREATE／INSERT 檢查 warnings，不能把截斷／轉型視為成功。 +4. staging catalog／結構／逐欄內容 SHA256 全部吻合、DML 提交後再檢查 staging 與來源，才送出**一個** multi-table RENAME。只限上述版本的 InnoDB 普通表、無 FK／trigger 等相依物件，且不覆蓋名稱。碰撞會失敗，不先刪除目標；發布後再次校驗才 `Verified`。 +5. staging 之前拒絕為 `Refused`。CREATE 曾送出後失敗／取消為 `StagingRemains`，列出本輪 staging→原名與階段；DML 可回滾，但表不會消失，列數只是處理進度。RENAME 曾送出後任何失敗／ack lost 為 `Uncertain`，不推測已發布或已撤銷。兩者禁止 UI 續寫／盲重試，不自動 DROP；Task 與 parent gate 保持到實際操作停止。RENAME 送出後取消不撤銷已提交操作，需等發布／驗證結果。 + +沒有逐表 checkpoint、自動 cleanup 或 durable 客戶端 resume journal。程序／引擎 crash 不在本輪實際驗證範圍；重啟後任何 staging 或發布物件都讓「必須空目標」預檢拒絕，需 DBA 核對實際 catalog 與本輪報告，不把名稱前綴當成自動刪除授權。不得把正常 RENAME 碰撞探測／注入 ack lost 擴大為實測斷電 crash recovery。 + +精確子集：有／無整數單欄或複合 PK（手動給值、最多 16 欄）、具名普通升冪 bigint 索引；BIGINT signed ↔ bigint、LONGTEXT ↔ text C、LONGBLOB ↔ bytea、NULL／NOT NULL。表／欄／索引名只接受小寫 ASCII 63 字元，索引在來源與目標名稱空間不得碰撞。MySQL `PRIMARY` 映射 PG 預設 `表名前58字元_pkey`；PG 自訂 PK 名稱無法保留為 MySQL PRIMARY，預檢拒絕。映射依來源引擎判斷：PG 自訂名稱恰為 `PRIMARY` 仍是明確名稱，PG→PG 保留並實測 `ON CONFLICT ON CONSTRAINT "PRIMARY"`，不誤當 MySQL 內建名稱。恰巧符合預設慣例的手動 PG 名稱無法由 catalog 區分；如依賴該 constraint 名稱，應用 PG 目標，不承諾跨引擎 ON CONFLICT 或字串函式行為相同。 + +文字／二進位**可以作一般欄位，但不能作此 MySQL 配對的索引鍵**：MySQL LONGTEXT／LONGBLOB 鍵需要 prefix，縮成 VARCHAR／VARBINARY 會新增長度約束；本輪不偷偷截鍵或改型別。unsigned／其他整數寬度、DECIMAL／時間／浮點、CHAR／VARCHAR／TEXT／BLOB 的其他長度、ENUM／SET、zero date、AUTO_INCREMENT、default、generated、FK／CHECK／UNIQUE、前綴／expression／descending／invisible／特殊索引、partition、view、sequence、trigger／routine／event、自訂表屬性或 comments 均以 catalog＋SHOW CREATE TABLE 完整 grammar 預檢拒絕，沒有靜默跳過。需要這些語意仍須後續專用轉換契約。 + +目前僅能證明直接 global／schema-level 權限的完整 catalog 可見性：SELECT、SHOW VIEW、TRIGGER、EVENT、CREATE ROUTINE、ALTER ROUTINE；角色／表級授權或缺權限會阻擋。不執行 SHOW GRANTS（MariaDB 可能含認證資訊）、不讀秘密或要求新增權限。 + +可重跑：`python3 tests/Run-MySqlLogicalMigration.py [新證據目錄]`，先快取官方 mysql:8.4、mariadb:11.4、postgres:16-alpine。直接引用正式 Core；唯一 ownership label／loopback 隨機 ports／tmpfs，finally 僅清自有容器。`migration.log`／`evidence.json` 保存實際版本、image digest、來源 hash／清理狀態，不保存資料值／密碼。CI Linux 另跑此 fixture，四平台編譯；這與既有 provider 56/56 分開。 + +官方語意依據：MySQL 8.4 [implicit commit](https://dev.mysql.com/doc/refman/8.4/en/implicit-commit.html)／[RENAME TABLE](https://dev.mysql.com/doc/refman/8.4/en/rename-table.html)，MariaDB [implicit commit](https://mariadb.com/docs/server/reference/sql-statements/transactions/sql-statements-that-cause-an-implicit-commit)／[RENAME TABLE](https://mariadb.com/docs/server/reference/sql-statements/data-definition/rename-table)。限於受控版本與物件子集，不宣稱所有 DDL 可回滾。 ## 可重跑的 PostgreSQL 跨版本證據 @@ -61,13 +85,13 @@ SQLite／PostgreSQL 工具入口在 Avalonia 工作台「邏輯搬移」。來 | 方向 | 現況／執行前拒絕的技術理由 | 下一階段必要能力（不是本輪已通過） | | --- | --- | --- | -| MySQL/MariaDB → SQLite/PG/SQLServer | 整數 unsigned 上界、zero date、ENUM/SET、charset、PAD SPACE 定序、AUTO_INCREMENT 與 storage engine 一致快照不能用現有 count-only transfer 當忠實轉換 | 來源 catalog／版本／InnoDB 快照 adapter，逐欄區間及 charset/定序證明；不可先沿用一般 type map 再補驗證 | -| SQLite/PG/SQLServer → MySQL/MariaDB，以及 MySQL↔MariaDB／跨版本 | MySQL DDL implicit commit，DDL＋資料單一交易會是假保證；目標即使原本空白，取消仍可能已留下表 | 另建 staging 與 durable uncertain/checkpoint 協議、預檢引擎版本差異、人工核對恢復；不能自動刪 staging 或覆蓋目標假裝 rollback | +| MySQL/MariaDB → SQLite/SQLServer | unsigned、zero date、ENUM/SET、charset、PAD SPACE、AUTO_INCREMENT 等仍無等價映射；與 PG 的受控子集已落地 | 另訂目的端 schema／資料 codec 與語意驗證，不把 count-only transfer 當忠實轉換 | +| SQLite/SQLServer → MySQL/MariaDB，以及 MySQL↔MariaDB／跨版本 | 不開放這些配對；與 PG 的 staging 路徑不等於其他來源語意已可保留 | 依目的端型別／定序契約與引擎版本驗證擴充；不直接沿用既有對映或自動刪 staging | | SQLServer → SQLite/PG/MySQL，以及反向 | UTF-16/UTF-8、linguistic/BIN2 定序的比較／尾空白規則、rowversion、decimal/time scale、identity/default/sequence 不能把現有型別名稱替換當成保留語意 | 可重用 SQLServer catalog 的阻擋規則，但需跨 provider canonical schema／資料 codec、來源一致快照與逐欄語意／版本證據；目前只 Windows SQLServer→SQLServer 專用子集可執行 | | Oracle/Snowflake 與其他引擎 | NUMBER／時間、空字串 NULL、identifier/collation 與交易 DDL 不等價 | 專用 catalog／codec／交易恢復設計；本輪不開連線 | | MongoDB、Redis/Garnet 與表格引擎 | document/key-value model、TTL／原子操作／巢狀內容沒有通用無損表格映射 | 明確專用模型轉換契約，不當成同義 restore | -上述方向服務在開連線前拒絕，沒有聲稱某個 MySQL/MariaDB 或 SQLServer 版本已跨引擎通過。雙向跨版本／換引擎的整體需求仍未全部完成；此批是實際擴充 SQLite/PG 的可驗證第二階段，而不是全引擎完成或發版的宣告。 +上述其他方向服務在開連線前拒絕。雙向跨版本／換引擎的整體需求仍未全部完成；目前已擴充 SQLite/PG 與 MySQL/MariaDB↔PG 的可驗證子集，不是全引擎完成或發版宣告。release 仍 hold。 有 FK、其他字串定序、時間、decimal、identity、procedure/trigger 的一般資料庫仍須使用引擎官方 schema/data migration 工具與人工轉換／驗證，不是宣稱 bacpac 可無限制降版。新版 SQL Server .bak 仍不能直接還原到舊引擎，compatibility level 不改備份格式；既有原生 restore 預檢繼續保留。 @@ -79,4 +103,4 @@ SQLite／PostgreSQL 工具入口在 Avalonia 工作台「邏輯搬移」。來 Avalonia/Xvfb：正式入口持有主視窗 gate、預覽、目標變更失效、錯誤確認不寫入、實際確認寫入且禁止重複 resume。既有跨平台 CI fixture 另測 SQLite ↔ PostgreSQL 16，包括正式 catalog 逐項加 feature／移除 feature 後恢復可預檢。 -真 SQL Server 跨版本、SQL auth+SSH 真機、PostgreSQL 14/15/17、SQLite 歷史版本及非關聯式轉換皆沒有完整矩陣證據；不得把 mock／同版本一般 provider round-trip 當全部資料庫版本已通過。 +真 SQL Server 2019↔2022 與 PG14↔17／16↔17 的限定 fixture 已有證據；SQL auth+SSH 真機、PG15、MySQL/MariaDB 的其他版本配對、SQLite 歷史版本及非關聯式轉換皆沒有完整矩陣證據。不得把 mock／同版本一般 provider round-trip 當全部資料庫版本已通過。 diff --git a/mySQLPunk.Core/Models/LogicalMigration.cs b/mySQLPunk.Core/Models/LogicalMigration.cs index 022a443..583234f 100644 --- a/mySQLPunk.Core/Models/LogicalMigration.cs +++ b/mySQLPunk.Core/Models/LogicalMigration.cs @@ -16,8 +16,8 @@ public sealed record LogicalMigrationPreview( public bool CanStart => Blockers.Count == 0 && Tables.Count > 0; } -public enum LogicalMigrationOutcome { Verified, RolledBack, Uncertain, Refused } +public enum LogicalMigrationOutcome { Verified, RolledBack, Uncertain, Refused, StagingRemains } public sealed record LogicalMigrationResult(LogicalMigrationOutcome Outcome, long Rows, string Report) { - public bool RetryBlocked => Outcome is LogicalMigrationOutcome.Uncertain or LogicalMigrationOutcome.Verified; + public bool RetryBlocked => Outcome is LogicalMigrationOutcome.Uncertain or LogicalMigrationOutcome.Verified or LogicalMigrationOutcome.StagingRemains; } diff --git a/mySQLPunk.Core/Services/LogicalMigrationKeys.cs b/mySQLPunk.Core/Services/LogicalMigrationKeys.cs index b10388b..a150ae9 100644 --- a/mySQLPunk.Core/Services/LogicalMigrationKeys.cs +++ b/mySQLPunk.Core/Services/LogicalMigrationKeys.cs @@ -28,9 +28,12 @@ internal static bool PostgreSqlIndexKeyFits(int blockSize, IReadOnlyList table[..Math.Min(table.Length, 58)] + "_pkey"; - private static LogicalMigrationTable TargetTable(DatabaseProviderKind provider, LogicalMigrationTable table) => + private static LogicalMigrationTable TargetTable(DatabaseProviderKind provider, LogicalMigrationTable table, + DatabaseProviderKind? sourceProvider = null) => table with { PrimaryKeyName = table.PrimaryKey.Count == 0 || provider == DatabaseProviderKind.Sqlite - ? null : table.PrimaryKeyName ?? DefaultPrimaryKeyName(table.Name) }; + ? null : provider == DatabaseProviderKind.MySql ? "PRIMARY" : table.PrimaryKeyName == null || + sourceProvider == DatabaseProviderKind.MySql && table.PrimaryKeyName == "PRIMARY" + ? DefaultPrimaryKeyName(table.Name) : table.PrimaryKeyName }; private sealed record Keys(IReadOnlyList PrimaryKey, IReadOnlyList Indexes, string? PrimaryKeyName); diff --git a/mySQLPunk.Core/Services/LogicalMigrationMySql.cs b/mySQLPunk.Core/Services/LogicalMigrationMySql.cs new file mode 100644 index 0000000..0460d60 --- /dev/null +++ b/mySQLPunk.Core/Services/LogicalMigrationMySql.cs @@ -0,0 +1,355 @@ +using System.Data; +using System.Data.Common; +using System.Globalization; +using System.Text; +using System.Text.Json; +using System.Text.RegularExpressions; +using MySqlConnector; +using MySqlPunk.Core.Models; +using MySqlPunk.Core.Providers; + +namespace MySqlPunk.Core.Services; + +public static partial class LogicalMigrationService +{ + internal static Task RunWithMySqlObserverAsync(ConnectionProfile source, ConnectionProfile target, + int selectedTargetMajor, LogicalMigrationPreview approved, CancellationToken token, Action observer) => + Task.Run(() => RunCoreAsync(source.Clone(), target.Clone(), selectedTargetMajor, approved, target.Database, token, null, observer), token); + + private static IsolationLevel SourceIsolation(DatabaseProviderKind provider) => provider switch + { + DatabaseProviderKind.Sqlite => IsolationLevel.Serializable, + DatabaseProviderKind.MySql => IsolationLevel.RepeatableRead, + _ => IsolationLevel.ReadCommitted + }; + + private static string MySqlQuote(string name) => "`" + name.Replace("`", "``") + "`"; + private static bool IsMariaDb(string version) => version.Contains("MariaDB", StringComparison.Ordinal); + private static string MySqlCollation(string version) => IsMariaDb(version) ? "utf8mb4_nopad_bin" : "utf8mb4_0900_bin"; + + private static void ValidateMySqlDirection(ConnectionProfile source, ConnectionProfile target, Snapshot origin, Snapshot destination, List blockers) + { + if (source.Provider != DatabaseProviderKind.PostgreSql && target.Provider != DatabaseProviderKind.PostgreSql) + blockers.Add("MySQL／MariaDB 本階段只允許與 PostgreSQL 雙向邏輯搬移，不開放其他配對。"); + if (origin.Tables.Count > 100) blockers.Add("MySQL 配對最多 100 表，限制單次 staging 與 RENAME 語句規模。"); + if (target.Provider == DatabaseProviderKind.MySql && + (destination.PacketBytes < 65536 || origin.Data.Values.SelectMany(rows => rows).Any(row => MySqlRowBytes(row) + 16384 > MySqlBatchBytes(destination.PacketBytes)))) + blockers.Add("資料列超出目標 max_allowed_packet 的保守封包預算（含 UTF-8／binary escaping），不自動修改伺服器設定。"); + foreach (var table in origin.Tables) + { + if (!Regex.IsMatch(table.Name, "\\A[a-z_][a-z0-9_]{0,62}\\z") || table.Columns.Any(column => !Regex.IsMatch(column.Name, "\\A[a-z_][a-z0-9_]{0,62}\\z")) || + table.Indexes.Any(index => !Regex.IsMatch(index.Name, "\\A[a-z_][a-z0-9_]{0,62}\\z"))) + blockers.Add(table.Name + "：MySQL 配對只允許小寫 ASCII 名稱，避免大小寫配置差異。"); + var keys = table.Indexes.Select(index => index.Columns).Concat(table.PrimaryKey.Count == 0 ? Array.Empty>() : new[] { table.PrimaryKey }); + if (keys.Any(key => key.Count is < 1 or > 16 || key.Any(name => table.Columns.All(column => column.Name != name || column.Kind != "integer")))) + blockers.Add(table.Name + "：MySQL 配對只保留最多 16 欄 signed bigint 主鍵／普通索引；TEXT／BLOB 前綴索引不能忠實替代完整鍵。"); + if (target.Provider == DatabaseProviderKind.MySql && table.PrimaryKeyName != null && table.PrimaryKeyName != DefaultPrimaryKeyName(table.Name)) + blockers.Add(table.Name + "：MySQL 的 PRIMARY 無法保留自訂 PostgreSQL constraint 名稱。"); + } + if (origin.Tables.SelectMany(table => table.Indexes).Select(index => index.Name).GroupBy(name => name, StringComparer.Ordinal).Any(group => group.Count() > 1)) + blockers.Add("MySQL 表內索引名稱在 PostgreSQL 全域名稱空間重複，不能靜默更名。"); + } + + private static string CreateMySqlTableSql(LogicalMigrationTable table, string collation) => + "CREATE TABLE " + MySqlQuote(table.Name) + " (" + string.Join(",", table.Columns.Select(column => + MySqlQuote(column.Name) + " " + (column.Kind == "integer" ? "bigint" : column.Kind == "binary" ? "longblob" : "longtext CHARACTER SET utf8mb4 COLLATE " + collation) + + (column.Nullable ? " DEFAULT NULL" : " NOT NULL"))) + + (table.PrimaryKey.Count == 0 ? "" : ",PRIMARY KEY (" + string.Join(",", table.PrimaryKey.Select(MySqlQuote)) + ")") + + string.Concat(table.Indexes.Select(index => ",KEY " + MySqlQuote(index.Name) + " (" + string.Join(",", index.Columns.Select(MySqlQuote)) + ")")) + + ") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=" + collation + ";"; + + private static async Task ReadMySqlSnapshotAsync(DbConnection connection, DbTransaction? transaction, bool readData, CancellationToken token) + { + var blockers = new List(); + var version = Convert.ToString(await ScalarAsync(connection, transaction, "SELECT VERSION();", token)) ?? ""; + var packetBytes = Convert.ToInt64(await ScalarAsync(connection, transaction, "SELECT @@max_allowed_packet;", token)); + var validVersion = IsMariaDb(version) ? Regex.IsMatch(version, "\\A11\\.4\\.[0-9]+-MariaDB") : Regex.IsMatch(version, "\\A8\\.4\\.[0-9]+\\z"); + if (!validVersion) blockers.Add("只驗證 MySQL 8.4.x／MariaDB 11.4.x；其他相容伺服器或版本不放行。"); + var collation = MySqlCollation(version); + if (Convert.ToInt64(await ScalarAsync(connection, transaction, "SELECT @@lower_case_table_names;", token)) != 0) + blockers.Add("要求 lower_case_table_names=0；不更改伺服器設定。"); + var identity = Convert.ToString(await ScalarAsync(connection, transaction, "SELECT CONCAT(DATABASE(),':',@@hostname,':',@@port);", token)) ?? ""; + if (string.IsNullOrEmpty(identity)) blockers.Add("無法確認實際資料庫識別。"); + var encoding = Convert.ToString(await ScalarAsync(connection, transaction, + "SELECT CONCAT(DEFAULT_CHARACTER_SET_NAME,':',DEFAULT_COLLATION_NAME) FROM information_schema.SCHEMATA WHERE SCHEMA_NAME=DATABASE();", token)); + if (encoding != "utf8mb4:" + collation) blockers.Add("資料庫必須使用 utf8mb4／" + collation + "，不自動修改 encoding／collation。"); + var visiblePrivileges = Convert.ToInt64(await ScalarAsync(connection, transaction, """ + SELECT COUNT(DISTINCT PRIVILEGE_TYPE) FROM ( + SELECT PRIVILEGE_TYPE FROM information_schema.USER_PRIVILEGES + WHERE GRANTEE=CONCAT(CHAR(39),SUBSTRING_INDEX(CURRENT_USER(),'@',1),CHAR(39),'@',CHAR(39),SUBSTRING_INDEX(CURRENT_USER(),'@',-1),CHAR(39)) + UNION ALL SELECT PRIVILEGE_TYPE FROM information_schema.SCHEMA_PRIVILEGES + WHERE TABLE_SCHEMA=DATABASE() AND GRANTEE=CONCAT(CHAR(39),SUBSTRING_INDEX(CURRENT_USER(),'@',1),CHAR(39),'@',CHAR(39),SUBSTRING_INDEX(CURRENT_USER(),'@',-1),CHAR(39)) + ) privileges WHERE PRIVILEGE_TYPE IN ('SELECT','SHOW VIEW','TRIGGER','EVENT','CREATE ROUTINE','ALTER ROUTINE'); + """, token)); + if (visiblePrivileges != 6) blockers.Add("無法證明目前帳號可完整看見 view／trigger／event／routine catalog（角色／表級授權未支援）；不讀秘密、不增授權。"); + var unsupported = await ScalarAsync(connection, transaction, """ + SELECT + (SELECT COUNT(*) FROM information_schema.TABLES WHERE TABLE_SCHEMA=DATABASE() AND + (TABLE_TYPE<>'BASE TABLE' OR ENGINE<>'InnoDB' OR ROW_FORMAT<>'Dynamic' OR CREATE_OPTIONS<>'' OR TABLE_COMMENT<>'')) + + (SELECT COUNT(*) FROM information_schema.TRIGGERS WHERE TRIGGER_SCHEMA=DATABASE()) + + (SELECT COUNT(*) FROM information_schema.ROUTINES WHERE ROUTINE_SCHEMA=DATABASE()) + + (SELECT COUNT(*) FROM information_schema.EVENTS WHERE EVENT_SCHEMA=DATABASE()) + + (SELECT COUNT(*) FROM information_schema.PARTITIONS WHERE TABLE_SCHEMA=DATABASE() AND PARTITION_NAME IS NOT NULL) + + (SELECT COUNT(*) FROM information_schema.TABLE_CONSTRAINTS WHERE CONSTRAINT_SCHEMA=DATABASE() AND CONSTRAINT_TYPE<>'PRIMARY KEY'); + """, token); + if (Convert.ToInt64(unsupported) != 0) blockers.Add("存在非 InnoDB Dynamic 普通表、view、sequence、trigger、routine、event、partition、FK／CHECK／UNIQUE 或自訂表屬性；不能略過。"); + var names = new List(); + await using (var command = connection.CreateCommand()) + { + command.Transaction = transaction; + command.CommandText = "SELECT TABLE_NAME FROM information_schema.TABLES WHERE TABLE_SCHEMA=DATABASE() ORDER BY BINARY TABLE_NAME;"; + await using var reader = await command.ExecuteReaderAsync(token).ConfigureAwait(false); + while (await reader.ReadAsync(token).ConfigureAwait(false)) names.Add(reader.GetString(0)); + } + var tables = new List(); + var data = new Dictionary>(); + long totalRows = 0, totalBytes = 0; + foreach (var name in names) + { + if (!Identifier.IsMatch(name)) { blockers.Add("表名不在安全子集。"); continue; } + var columns = new List(); + await using (var command = connection.CreateCommand()) + { + command.Transaction = transaction; + command.CommandText = "SELECT COLUMN_NAME,COLUMN_TYPE,IS_NULLABLE,COLUMN_DEFAULT,EXTRA,COLLATION_NAME,GENERATION_EXPRESSION,COLUMN_COMMENT FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=@table ORDER BY ORDINAL_POSITION;"; + var parameter = command.CreateParameter(); parameter.ParameterName = "@table"; parameter.Value = name; command.Parameters.Add(parameter); + await using var reader = await command.ExecuteReaderAsync(token).ConfigureAwait(false); + while (await reader.ReadAsync(token).ConfigureAwait(false)) + { + var type = reader.GetString(1); + var kind = type is "bigint" or "bigint(20)" ? "integer" : type == "longblob" ? "binary" : type == "longtext" ? "text" : "unsupported"; + if (!Identifier.IsMatch(reader.GetString(0)) || kind == "unsupported" || (!reader.IsDBNull(3) && !(IsMariaDb(version) && reader.GetString(3) == "NULL")) || reader.GetString(4) != "" || + (!reader.IsDBNull(6) && reader.GetString(6) != "") || reader.GetString(7) != "" || + (kind == "text" ? reader.IsDBNull(5) || reader.GetString(5) != collation : !reader.IsDBNull(5))) + blockers.Add(name + "/" + reader.GetString(0) + "：只接受 signed BIGINT／LONGTEXT NO PAD binary／LONGBLOB，無 default／generated／identity／自訂欄位屬性。"); + columns.Add(new(reader.GetString(0), kind, reader.GetString(2) == "YES")); + } + } + var primary = new List(); + var indexes = new SortedDictionary>(StringComparer.Ordinal); + await using (var command = connection.CreateCommand()) + { + command.Transaction = transaction; + command.CommandText = "SELECT INDEX_NAME,COLUMN_NAME,NON_UNIQUE,INDEX_TYPE,SUB_PART,COLLATION,INDEX_COMMENT," + + (IsMariaDb(version) ? "IGNORED" : "IS_VISIBLE") + " FROM information_schema.STATISTICS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=@table ORDER BY BINARY INDEX_NAME,SEQ_IN_INDEX;"; + var parameter = command.CreateParameter(); parameter.ParameterName = "@table"; parameter.Value = name; command.Parameters.Add(parameter); + await using var reader = await command.ExecuteReaderAsync(token).ConfigureAwait(false); + while (await reader.ReadAsync(token).ConfigureAwait(false)) + { + var indexName = reader.GetString(0); + if (reader.IsDBNull(1) || (indexName != "PRIMARY" && !Identifier.IsMatch(indexName)) || + Convert.ToInt32(reader.GetValue(2)) != (indexName == "PRIMARY" ? 0 : 1) || reader.GetString(3) != "BTREE" || + !reader.IsDBNull(4) || reader.IsDBNull(5) || reader.GetString(5) != "A" || reader.GetString(6) != "" || + reader.GetString(7) != (IsMariaDb(version) ? "NO" : "YES")) + { blockers.Add(name + "：不支援 unique／prefix／expression／descending／invisible 或非 BTREE 索引。"); continue; } + if (indexName == "PRIMARY") primary.Add(reader.GetString(1)); + else + { + if (!indexes.TryGetValue(indexName, out var fields)) indexes.Add(indexName, fields = new()); + fields.Add(reader.GetString(1)); + } + } + } + if (columns.Count is < 1 or > 100 || primary.Any(field => columns.All(column => column.Name != field || column.Nullable))) + blockers.Add(name + "：欄位數／主鍵 NOT NULL 不符。"); + var shape = new LogicalMigrationTable(name, columns, 0, "") + { + PrimaryKey = primary, + Indexes = indexes.Select(index => new LogicalMigrationIndex(index.Key, index.Value)).ToArray() + }; + await using (var command = connection.CreateCommand()) + { + command.Transaction = transaction; command.CommandText = "SHOW CREATE TABLE " + MySqlQuote(name); + await using var reader = await command.ExecuteReaderAsync(token).ConfigureAwait(false); + if (!await reader.ReadAsync(token).ConfigureAwait(false) || !SafeMySqlDefinition(shape, reader.GetString(1), collation)) + blockers.Add(name + "：SHOW CREATE TABLE 包含不支援語意或 catalog 與 DDL 不符。"); + } + var rows = new List(); + if (readData && blockers.Count == 0) + { + await using var command = connection.CreateCommand(); command.Transaction = transaction; + command.CommandText = "SELECT " + string.Join(",", columns.Select(column => + { + var quoted = MySqlQuote(column.Name); + if (column.Kind == "integer") return quoted; + var length = "OCTET_LENGTH(" + quoted + ")"; + return length + ",CASE WHEN " + length + "<=" + MaximumFieldBytes + " THEN CAST(" + quoted + " AS BINARY) ELSE NULL END"; + })) + " FROM " + MySqlQuote(name) + " LOCK IN SHARE MODE;"; + await using var reader = await command.ExecuteReaderAsync(CommandBehavior.SequentialAccess, token).ConfigureAwait(false); + while (await reader.ReadAsync(token).ConfigureAwait(false)) + { + if (++totalRows > MaximumRows) throw new InvalidOperationException("來源超過 100,000 列。"); + var row = new object?[columns.Count]; + var ordinal = 0; + for (var columnIndex = 0; columnIndex < columns.Count; columnIndex++) + { + var binary = columns[columnIndex].Kind != "integer"; + var bytes = reader.IsDBNull(ordinal) ? 8 : binary ? Convert.ToInt64(reader.GetValue(ordinal), CultureInfo.InvariantCulture) : 8; + totalBytes += Math.Max(8, bytes); + if (bytes > MaximumFieldBytes || totalBytes > MaximumBytes) throw new InvalidOperationException("超過單欄 1 MiB／總量 64 MiB 預覽上限。"); + if (binary) ordinal++; + object? value = reader.IsDBNull(ordinal) ? null : binary ? ReadBoundedBytes(reader, ordinal, bytes, token) : reader.GetInt64(ordinal); + if (value is byte[] encoded && columns[columnIndex].Kind == "text") + { + value = new UTF8Encoding(false, true).GetString(encoded); + if (((string)value).Contains('\0')) throw new InvalidOperationException("文字含 NUL，PostgreSQL 無法保存。"); + } + ordinal++; row[columnIndex] = value; + } + rows.Add(row); + } + } + tables.Add(new(name, columns, rows.Count, ContentHash(rows)) + { + PrimaryKey = primary, PrimaryKeyName = primary.Count == 0 ? null : "PRIMARY", + Indexes = indexes.Select(index => new LogicalMigrationIndex(index.Key, index.Value)).ToArray() + }); + data.Add(name, rows); + } + return new(version, identity, tables, data, blockers, 0, packetBytes); + } + + private static long MySqlBatchBytes(long packetBytes) => Math.Min(packetBytes / 2, 4 * 1024 * 1024); + private static long MySqlRowBytes(object?[] row) => row.Sum(value => 64L + (value is byte[] binary ? 2L * binary.Length : value is string text ? 2L * Encoding.UTF8.GetByteCount(text) : 32)); + + private static IEnumerable MySqlBatches(List rows, long packetBytes) + { + var batch = new List(); + long bytes = 16384; + foreach (var row in rows) + { + var size = MySqlRowBytes(row); + if (batch.Count > 0 && (batch.Count == 100 || bytes + size > MySqlBatchBytes(packetBytes))) + { yield return batch.ToArray(); batch.Clear(); bytes = 16384; } + batch.Add(row); bytes += size; + } + if (batch.Count > 0) yield return batch.ToArray(); + } + + private static bool SafeMySqlDefinition(LogicalMigrationTable table, string sql, string collation) + { + var columns = table.Columns.Select(column => Regex.Escape(MySqlQuote(column.Name)) + "\\s+" + + (column.Kind == "integer" ? "bigint(?:\\(20\\))?" : column.Kind == "binary" ? "longblob" : + "longtext(?:\\s+CHARACTER\\s+SET\\s+utf8mb4)?(?:\\s+COLLATE\\s+" + collation + ")?") + + (column.Nullable ? "(?:\\s+DEFAULT\\s+NULL)?" : "\\s+NOT\\s+NULL")); + var key = table.PrimaryKey.Count == 0 ? "" : "\\s*,\\s*PRIMARY\\s+KEY\\s*\\(\\s*" + string.Join("\\s*,\\s*", table.PrimaryKey.Select(name => Regex.Escape(MySqlQuote(name)))) + "\\s*\\)"; + var indexes = table.Indexes.Select(index => "\\s*,\\s*KEY\\s+" + Regex.Escape(MySqlQuote(index.Name)) + "\\s*\\(\\s*" + string.Join("\\s*,\\s*", index.Columns.Select(name => Regex.Escape(MySqlQuote(name)))) + "\\s*\\)"); + return Regex.IsMatch(sql, "\\ACREATE\\s+TABLE\\s+" + Regex.Escape(MySqlQuote(table.Name)) + "\\s*\\(\\s*" + + string.Join("\\s*,\\s*", columns) + key + string.Concat(indexes) + "\\s*\\)\\s+ENGINE=InnoDB\\s+DEFAULT\\s+CHARSET=utf8mb4\\s+COLLATE=" + collation + "\\z", + RegexOptions.IgnoreCase | RegexOptions.CultureInvariant); + } + + private static async Task RunMySqlTargetAsync(ConnectionProfile source, ConnectionProfile target, + int selectedTargetMajor, LogicalMigrationPreview approved, CancellationToken token, Action? observer) + { + using var sourceOwner = (AdoDatabaseSession)DatabaseProviderFactory.Create(source); + using var targetOwner = (AdoDatabaseSession)DatabaseProviderFactory.Create(target); + await using var origin = await sourceOwner.OpenMigrationConnectionAsync(source.Database, token, readOnly: true).ConfigureAwait(false); + await using var destination = await targetOwner.OpenMigrationConnectionAsync(target.Database, token).ConfigureAwait(false); + await using var sourceTransaction = await origin.BeginTransactionAsync(SourceIsolation(source.Provider), token).ConfigureAwait(false); + var staging = new Dictionary(StringComparer.Ordinal); + var ddlAttempted = false; + var publishAttempted = false; + long rows = 0; + var stage = "重新預檢"; + try + { + var snapshot = await ReadSnapshotAsync(origin, sourceTransaction, source.Provider, true, token).ConfigureAwait(false); + var targetSnapshot = await ReadMySqlSnapshotAsync(destination, null, false, token).ConfigureAwait(false); + var current = BuildPreview(source, target, selectedTargetMajor, snapshot, targetSnapshot); + if (!current.CanStart || current.Fingerprint != approved.Fingerprint) + return new(LogicalMigrationOutcome.Refused, 0, "來源／目標與確認快照不符；未寫入。"); + var prefix = "_mp_" + Guid.NewGuid().ToString("N") + "_"; + for (var index = 0; index < snapshot.Tables.Count; index++) staging.Add(prefix + index, snapshot.Tables[index].Name); + stage = "建立本輪 staging（DDL 隱含提交）"; + foreach (var entry in staging) + { + token.ThrowIfCancellationRequested(); ddlAttempted = true; + var table = snapshot.Tables.Single(table => table.Name == entry.Value) with { Name = entry.Key }; + await ExecuteAsync(destination, null, CreateMySqlTableSql(table, MySqlCollation(targetSnapshot.Version)), token).ConfigureAwait(false); + await EnsureNoMySqlWarningsAsync(destination, null, token).ConfigureAwait(false); + observer?.Invoke("staging", destination, null); + } + stage = "staging 分批 DML 交易"; + await using (var transaction = await destination.BeginTransactionAsync(IsolationLevel.RepeatableRead, token).ConfigureAwait(false)) + { + foreach (var entry in staging) + { + var table = snapshot.Tables.Single(table => table.Name == entry.Value); + foreach (var batch in MySqlBatches(snapshot.Data[table.Name], targetSnapshot.PacketBytes)) + { + token.ThrowIfCancellationRequested(); + await using var command = destination.CreateCommand(); command.Transaction = transaction; + var values = new List(); + foreach (var row in batch) + { + var names = new List(); + for (var columnIndex = 0; columnIndex < row.Length; columnIndex++) + { + var parameter = command.CreateParameter(); parameter.ParameterName = "@p" + command.Parameters.Count; + parameter.DbType = table.Columns[columnIndex].Kind == "integer" ? DbType.Int64 : table.Columns[columnIndex].Kind == "text" ? DbType.String : DbType.Binary; + parameter.Value = row[columnIndex] ?? DBNull.Value; + names.Add(parameter.ParameterName); command.Parameters.Add(parameter); + } + values.Add("(" + string.Join(",", names) + ")"); + } + command.CommandText = "INSERT INTO " + MySqlQuote(entry.Key) + " (" + string.Join(",", table.Columns.Select(column => MySqlQuote(column.Name))) + ") VALUES " + string.Join(",", values); + await command.ExecuteNonQueryAsync(token).ConfigureAwait(false); + await EnsureNoMySqlWarningsAsync(destination, transaction, token).ConfigureAwait(false); + rows += batch.Length; + observer?.Invoke("batch", destination, transaction); + } + } + stage = "staging 內容與結構校驗"; + observer?.Invoke("verify", destination, transaction); + var written = await ReadMySqlSnapshotAsync(destination, transaction, true, token).ConfigureAwait(false); + VerifyMySqlStaging(snapshot, written, staging); + token.ThrowIfCancellationRequested(); + await transaction.CommitAsync(CancellationToken.None).ConfigureAwait(false); + } + stage = "發布前確認來源與 staging"; + var finalSource = await ReadSnapshotAsync(origin, sourceTransaction, source.Provider, true, token).ConfigureAwait(false); + if (finalSource.Blockers.Count != 0 || JsonSerializer.Serialize(finalSource.Tables) != JsonSerializer.Serialize(snapshot.Tables)) + throw new InvalidOperationException("來源在搬移期間改變。"); + await using (var transaction = await destination.BeginTransactionAsync(IsolationLevel.RepeatableRead, token).ConfigureAwait(false)) + { + VerifyMySqlStaging(snapshot, await ReadMySqlSnapshotAsync(destination, transaction, true, token).ConfigureAwait(false), staging); + await transaction.RollbackAsync(CancellationToken.None).ConfigureAwait(false); + } + stage = "單一 multi-table RENAME 發布（非 ROLLBACK）"; + observer?.Invoke("publish", destination, null); + token.ThrowIfCancellationRequested(); publishAttempted = true; + await ExecuteAsync(destination, null, "RENAME TABLE " + string.Join(",", staging.Select(entry => MySqlQuote(entry.Key) + " TO " + MySqlQuote(entry.Value))) + ";", CancellationToken.None).ConfigureAwait(false); + observer?.Invoke("published", destination, null); + await using (var transaction = await destination.BeginTransactionAsync(IsolationLevel.RepeatableRead, CancellationToken.None).ConfigureAwait(false)) + { + var published = await ReadMySqlSnapshotAsync(destination, transaction, true, CancellationToken.None).ConfigureAwait(false); + VerifyMySqlStaging(snapshot, published, snapshot.Tables.ToDictionary(table => table.Name, table => table.Name)); + await transaction.RollbackAsync(CancellationToken.None).ConfigureAwait(false); + } + return new(LogicalMigrationOutcome.Verified, rows, "InnoDB staging 已驗證並以單一 RENAME TABLE 發布，發布後 schema／內容再次相符。DDL 不是可回滾交易；不自動重試。發布送出後不丟棄 Task，取消需等待實際結果。"); + } + catch (Exception exception) + { + return new(publishAttempted ? LogicalMigrationOutcome.Uncertain : ddlAttempted ? LogicalMigrationOutcome.StagingRemains : LogicalMigrationOutcome.Refused, + rows, "階段：" + stage + "；" + exception.GetType().Name + (token.IsCancellationRequested ? "；已取消" : "") + + "；列數為處理進度，不是已提交列數；沒有刪除或覆蓋物件，沒有聲稱 DDL 回滾。" + (publishAttempted ? "發布結果不確定，禁止重試。" : "本輪 staging 可能殘留，禁止續寫／盲重試。") + + "請 DBA 核對目標 " + target.Database + " 與本輪物件：" + string.Join(",", staging.Select(entry => entry.Key + " → " + entry.Value))); + } + } + + private static void VerifyMySqlStaging(Snapshot expected, Snapshot actual, IReadOnlyDictionary staging) + { + if (actual.Blockers.Count != 0 || actual.Tables.Any(table => !staging.ContainsKey(table.Name))) + throw new InvalidOperationException("staging catalog 有阻擋或出現其他物件。"); + var normalized = actual.Tables.Select(table => table with { Name = staging[table.Name] }).OrderBy(table => table.Name, StringComparer.Ordinal); + if (JsonSerializer.Serialize(normalized) != JsonSerializer.Serialize(expected.Tables.Select(table => TargetTable(DatabaseProviderKind.MySql, table)).OrderBy(table => table.Name, StringComparer.Ordinal))) + throw new InvalidOperationException("staging／發布目標 schema 或 SHA256 內容不符。"); + } + + private static async Task EnsureNoMySqlWarningsAsync(DbConnection connection, DbTransaction? transaction, CancellationToken token) + { + if (Convert.ToInt64(await ScalarAsync(connection, transaction, "SHOW COUNT(*) WARNINGS;", token)) != 0) + throw new InvalidOperationException("MySQL 回報 warnings，不能把轉型／截斷當成功。"); + } +} diff --git a/mySQLPunk.Core/Services/LogicalMigrationService.cs b/mySQLPunk.Core/Services/LogicalMigrationService.cs index b97a53e..df81781 100644 --- a/mySQLPunk.Core/Services/LogicalMigrationService.cs +++ b/mySQLPunk.Core/Services/LogicalMigrationService.cs @@ -18,7 +18,7 @@ public static partial class LogicalMigrationService private const long MaximumBytes = 64 * 1024 * 1024; private static readonly Regex Identifier = new("\\A[A-Za-z_][A-Za-z0-9_]{0,62}\\z", RegexOptions.CultureInvariant); private sealed record Snapshot(string Version, string Identity, List Tables, - Dictionary> Data, List Blockers, int BlockSize); + Dictionary> Data, List Blockers, int BlockSize, long PacketBytes = 0); public static Task PreviewAsync(ConnectionProfile source, ConnectionProfile target, int selectedTargetMajor, CancellationToken cancellationToken = default) => @@ -35,7 +35,12 @@ internal static Task RunWithObserverAsync(ConnectionProf Task.Run(() => RunCoreAsync(source.Clone(), target.Clone(), selectedTargetMajor, approved, confirmation, token, observer), token); private static bool Supported(ConnectionProfile profile) => - profile.Provider is DatabaseProviderKind.Sqlite or DatabaseProviderKind.PostgreSql; + profile.Provider is DatabaseProviderKind.Sqlite or DatabaseProviderKind.PostgreSql or DatabaseProviderKind.MySql; + + private static bool SupportedPair(ConnectionProfile source, ConnectionProfile target) => Supported(source) && Supported(target) && + (source.Provider != DatabaseProviderKind.MySql && target.Provider != DatabaseProviderKind.MySql || + source.Provider == DatabaseProviderKind.MySql && target.Provider == DatabaseProviderKind.PostgreSql || + source.Provider == DatabaseProviderKind.PostgreSql && target.Provider == DatabaseProviderKind.MySql); private static LogicalMigrationPreview Refused(ConnectionProfile target, string reason) => new("未識別", "未識別", target.Database, Array.Empty(), Array.Empty(), new[] { reason }, ""); @@ -43,15 +48,15 @@ private static LogicalMigrationPreview Refused(ConnectionProfile target, string private static async Task PreviewCoreAsync(ConnectionProfile source, ConnectionProfile target, int selectedTargetMajor, CancellationToken token) { - if (!Supported(source) || !Supported(target)) - return Refused(target, "可驗證跨引擎路徑為 SQLite STRICT ↔ PostgreSQL；MySQL 目標 DDL 隱含提交，SQL Server 定序／型別與 identity 需另行轉換預檢，不能沿用此交易保證或當成原生 restore。請見方向矩陣。"); + if (!SupportedPair(source, target)) + return Refused(target, "可驗證路徑為 SQLite STRICT ↔ PostgreSQL，以及 MySQL 8.4／MariaDB 11.4 ↔ PostgreSQL 受控子集；其他配對在開連線前拒絕，不當成原生 restore。請見方向矩陣。"); using var sourceOwner = (AdoDatabaseSession)DatabaseProviderFactory.Create(source); using var targetOwner = (AdoDatabaseSession)DatabaseProviderFactory.Create(target); await using var origin = await sourceOwner.OpenMigrationConnectionAsync(source.Database, token, readOnly: true).ConfigureAwait(false); await using var destination = await targetOwner.OpenMigrationConnectionAsync(target.Database, token).ConfigureAwait(false); using var sourceProgress = new SqliteProgress(origin, token); using var targetProgress = new SqliteProgress(destination, token); - await using var sourceTransaction = await origin.BeginTransactionAsync(source.Provider == DatabaseProviderKind.Sqlite ? IsolationLevel.Serializable : IsolationLevel.ReadCommitted, token).ConfigureAwait(false); + await using var sourceTransaction = await origin.BeginTransactionAsync(SourceIsolation(source.Provider), token).ConfigureAwait(false); try { var sourceSnapshot = await ReadSnapshotAsync(origin, sourceTransaction, source.Provider, true, token).ConfigureAwait(false); @@ -66,12 +71,14 @@ private static LogicalMigrationPreview BuildPreview(ConnectionProfile source, Co { var blockers = origin.Blockers.Select(reason => "來源：" + reason) .Concat(destination.Blockers.Select(reason => "目標：" + reason)).ToList(); + if (source.Provider == DatabaseProviderKind.MySql || target.Provider == DatabaseProviderKind.MySql) + ValidateMySqlDirection(source, target, origin, destination, blockers); if (target.Provider == DatabaseProviderKind.PostgreSql) { var relationNames = origin.Tables.Select(table => table.Name) .Concat(origin.Tables.SelectMany(table => table.Indexes.Select(index => index.Name))) .Concat(origin.Tables.Where(table => table.PrimaryKey.Count > 0) - .Select(table => TargetTable(target.Provider, table).PrimaryKeyName!)); + .Select(table => TargetTable(target.Provider, table, source.Provider).PrimaryKeyName!)); if (relationNames.GroupBy(name => name, StringComparer.Ordinal).Any(group => group.Count() > 1)) blockers.Add("表／普通索引／主鍵索引名稱在 PostgreSQL public 名稱空間衝突；不允許引擎自動更名。"); if (destination.BlockSize != 8192) blockers.Add("PostgreSQL 索引預算只驗證過 8192-byte block_size；未知／其他 page 大小不能放行。"); @@ -105,13 +112,15 @@ private static LogicalMigrationPreview BuildPreview(ConnectionProfile source, Co if (source.Provider == target.Provider && source.Database == target.Database && (source.Provider == DatabaseProviderKind.Sqlite || source.Host == target.Host && source.Port == target.Port)) blockers.Add("來源與目標不得為同一資料庫。"); - var statements = origin.Tables.Select(table => CreateTableSql(target.Provider, table)) - .Concat(origin.Tables.SelectMany(table => table.Indexes.Select(index => CreateIndexSql(target.Provider, table, index)))).ToArray(); + var statements = target.Provider == DatabaseProviderKind.MySql + ? origin.Tables.Select(table => CreateMySqlTableSql(table, MySqlCollation(destination.Version))).ToArray() + : origin.Tables.Select(table => CreateTableSql(target.Provider, TargetTable(target.Provider, table, source.Provider))) + .Concat(origin.Tables.SelectMany(table => table.Indexes.Select(index => CreateIndexSql(target.Provider, table, index)))).ToArray(); var fingerprint = Hash(JsonSerializer.Serialize(new { Source = new { source.Provider, source.Host, source.Port, source.Database }, Target = new { target.Provider, target.Host, target.Port, target.Database }, - origin.Version, origin.Identity, TargetIdentity = destination.Identity, TargetVersion = destination.Version, destination.BlockSize, origin.Tables, Statements = statements, Blockers = blockers + origin.Version, origin.Identity, TargetIdentity = destination.Identity, TargetVersion = destination.Version, destination.BlockSize, destination.PacketBytes, origin.Tables, Statements = statements, Blockers = blockers })); return new(origin.Version, destination.Version, target.Database, origin.Tables.AsReadOnly(), statements, blockers.AsReadOnly(), fingerprint); @@ -119,17 +128,19 @@ private static LogicalMigrationPreview BuildPreview(ConnectionProfile source, Co private static async Task RunCoreAsync(ConnectionProfile source, ConnectionProfile target, int selectedTargetMajor, LogicalMigrationPreview approved, string confirmation, CancellationToken token, - Action? observer) + Action? observer, Action? mySqlObserver = null) { - if (!approved.CanStart || confirmation != target.Database || !Supported(source) || !Supported(target)) + if (!approved.CanStart || confirmation != target.Database || !SupportedPair(source, target)) return new(LogicalMigrationOutcome.Refused, 0, "未確認完整目標名稱或預檢未通過；沒有寫入。"); + if (target.Provider == DatabaseProviderKind.MySql) + return await RunMySqlTargetAsync(source, target, selectedTargetMajor, approved, token, mySqlObserver).ConfigureAwait(false); using var sourceOwner = (AdoDatabaseSession)DatabaseProviderFactory.Create(source); using var targetOwner = (AdoDatabaseSession)DatabaseProviderFactory.Create(target); await using var origin = await sourceOwner.OpenMigrationConnectionAsync(source.Database, token, readOnly: true).ConfigureAwait(false); await using var destination = await targetOwner.OpenMigrationConnectionAsync(target.Database, token).ConfigureAwait(false); using var sourceProgress = new SqliteProgress(origin, token); using var targetProgress = new SqliteProgress(destination, token); - await using var sourceTransaction = await origin.BeginTransactionAsync(source.Provider == DatabaseProviderKind.Sqlite ? IsolationLevel.Serializable : IsolationLevel.ReadCommitted, token).ConfigureAwait(false); + await using var sourceTransaction = await origin.BeginTransactionAsync(SourceIsolation(source.Provider), token).ConfigureAwait(false); await using var targetTransaction = destination is SqliteConnection sqlite ? sqlite.BeginTransaction(IsolationLevel.Serializable, deferred: false) : await destination.BeginTransactionAsync(IsolationLevel.ReadCommitted, token).ConfigureAwait(false); @@ -189,7 +200,7 @@ private static async Task RunCoreAsync(ConnectionProfile stage = "逐欄型別、NULL、內容與結構驗證"; observer?.Invoke("verify", destination, targetTransaction); var written = await ReadSnapshotAsync(destination, targetTransaction, target.Provider, true, token).ConfigureAwait(false); - if (written.Blockers.Count != 0 || JsonSerializer.Serialize(written.Tables) != JsonSerializer.Serialize(sourceSnapshot.Tables.Select(table => TargetTable(target.Provider, table)))) + if (written.Blockers.Count != 0 || JsonSerializer.Serialize(written.Tables) != JsonSerializer.Serialize(sourceSnapshot.Tables.Select(table => TargetTable(target.Provider, table, source.Provider)))) throw new InvalidOperationException("目標結構或內容 SHA-256 多重集合驗證不符。"); var sourceFinal = await ReadSnapshotAsync(origin, sourceTransaction, source.Provider, true, token).ConfigureAwait(false); if (sourceFinal.Blockers.Count != 0 || JsonSerializer.Serialize(sourceFinal.Tables) != JsonSerializer.Serialize(sourceSnapshot.Tables)) @@ -220,6 +231,8 @@ private static async Task RunCoreAsync(ConnectionProfile private static async Task ReadSnapshotAsync(DbConnection connection, DbTransaction? transaction, DatabaseProviderKind provider, bool readData, CancellationToken token) { + if (provider == DatabaseProviderKind.MySql) + return await ReadMySqlSnapshotAsync(connection, transaction, readData, token).ConfigureAwait(false); var blockers = new List(); var blockSize = provider == DatabaseProviderKind.PostgreSql ? Convert.ToInt32(await ScalarAsync(connection, transaction, "SHOW block_size;", token), CultureInfo.InvariantCulture) : 0; @@ -404,7 +417,8 @@ private static bool SafeSqliteDefinition(string name, List + private static string CreateTableSql(DatabaseProviderKind provider, LogicalMigrationTable table) => provider == DatabaseProviderKind.MySql + ? CreateMySqlTableSql(table, "utf8mb4_0900_bin") : "CREATE TABLE " + Qualified(provider, table.Name) + " (" + string.Join(",", table.Columns.Select(column => Quote(column.Name) + " " + (column.Kind == "integer" ? provider == DatabaseProviderKind.Sqlite ? "INTEGER" : "bigint" : column.Kind == "text" ? provider == DatabaseProviderKind.Sqlite ? "TEXT COLLATE BINARY" : "text COLLATE \"C\"" diff --git a/mySQLPunk.Core/mySQLPunk.Core.csproj b/mySQLPunk.Core/mySQLPunk.Core.csproj index 76e1bc4..8fb6c66 100644 --- a/mySQLPunk.Core/mySQLPunk.Core.csproj +++ b/mySQLPunk.Core/mySQLPunk.Core.csproj @@ -10,6 +10,7 @@ + diff --git a/mySQLPunk.Desktop/LogicalMigrationWindow.cs b/mySQLPunk.Desktop/LogicalMigrationWindow.cs index fb956bb..7dea8ad 100644 --- a/mySQLPunk.Desktop/LogicalMigrationWindow.cs +++ b/mySQLPunk.Desktop/LogicalMigrationWindow.cs @@ -10,7 +10,7 @@ public sealed class LogicalMigrationWindow : Window { private readonly ConnectionProfile _source; private readonly ComboBox _target = new() { Name = "MigrationTarget", HorizontalAlignment = HorizontalAlignment.Stretch }; - private readonly ComboBox _version = new() { ItemsSource = new[] { "SQLite 3（3.37–3.53）", "PostgreSQL 14", "PostgreSQL 15", "PostgreSQL 16", "PostgreSQL 17" } }; + private readonly ComboBox _version = new() { ItemsSource = new[] { "SQLite 3（3.37–3.53）", "PostgreSQL 14", "PostgreSQL 15", "PostgreSQL 16", "PostgreSQL 17", "MySQL 8.4", "MariaDB 11.4" } }; private readonly TextBox _database = new() { Name = "MigrationDatabase", PlaceholderText = "另建空目標資料庫名稱／SQLite 完整檔案路徑" }; private readonly TextBox _password = new() { PasswordChar = '●', PlaceholderText = "目標密碼（只留記憶體，不讀秘密庫）" }; private readonly TextBox _sshPassword = new() { PasswordChar = '●', PlaceholderText = "目標 SSH 密碼（如需要）" }; @@ -30,7 +30,7 @@ public sealed class LogicalMigrationWindow : Window public LogicalMigrationWindow(ConnectionProfile source, IReadOnlyList targets, CancellationToken parentToken) { _source = source.Clone(); _parentToken = parentToken; - Title = "可驗證邏輯搬移（階段一）"; Width = 850; Height = 790; WindowStartupLocation = WindowStartupLocation.CenterOwner; + Title = "可驗證邏輯搬移（受控關聯式子集）"; Width = 850; Height = 790; WindowStartupLocation = WindowStartupLocation.CenterOwner; var profiles = targets.Select(profile => profile.Clone()).ToArray(); _target.ItemsSource = profiles.Select(profile => profile.Name + " · " + profile.ProviderDisplayName).ToArray(); var fields = new StackPanel { Spacing = 8, Margin = new Avalonia.Thickness(16) }; @@ -40,6 +40,7 @@ public LogicalMigrationWindow(ConnectionProfile source, IReadOnlyList { _preview = null; _apply.IsEnabled = false; @@ -53,7 +54,7 @@ public LogicalMigrationWindow(ConnectionProfile source, IReadOnlyList _version.SelectedIndex switch { 0 => 3, 5 => 8, 6 => 11, _ => 13 + _version.SelectedIndex }; } diff --git a/tests/DesktopProfileOperationHarness/Program.cs b/tests/DesktopProfileOperationHarness/Program.cs index dfb3361..d61e1aa 100644 --- a/tests/DesktopProfileOperationHarness/Program.cs +++ b/tests/DesktopProfileOperationHarness/Program.cs @@ -90,6 +90,12 @@ Check(window.OwnedWindows.Count() == 1 && ReferenceEquals(migrationCancellation, Field(window, "_operationCancellation")), "Actual migration dialog holds parent SQL/profile gate"); ((ComboBox)Field(migration, "_target")!).SelectedIndex = 0; ((ComboBox)Field(migration, "_version")!).SelectedIndex = 0; + var versions = (ComboBox)Field(migration, "_version")!; + versions.SelectedIndex = 5; + Check((int)Invoke(migration, "SelectedMajor")! == 8, "Actual migration UI selects MySQL 8.4 explicitly"); + versions.SelectedIndex = 6; + Check((int)Invoke(migration, "SelectedMajor")! == 11, "Actual migration UI selects MariaDB 11.4 explicitly"); + versions.SelectedIndex = 0; ((TextBox)Field(migration, "_database")!).Text = migrationTarget; var inspectMigration = (Button)Field(migration, "_inspect")!; var applyMigration = (Button)Field(migration, "_apply")!; diff --git a/tests/MySqlLogicalMigrationHarness/Program.cs b/tests/MySqlLogicalMigrationHarness/Program.cs new file mode 100644 index 0000000..e8f8b2d --- /dev/null +++ b/tests/MySqlLogicalMigrationHarness/Program.cs @@ -0,0 +1,249 @@ +using MySqlConnector; +using Npgsql; +using MySqlPunk.Core.Models; +using MySqlPunk.Core.Services; + +var password = Environment.GetEnvironmentVariable("MYSQLPUNK_TEST_PASSWORD") ?? throw new Exception("Missing fixture password."); +var pgPort = int.Parse(args[0]); +var myPorts = new[] { int.Parse(args[1]), int.Parse(args[2]) }; +string PgConnection(string database) => new NpgsqlConnectionStringBuilder +{ + Host = "127.0.0.1", Port = pgPort, Username = "postgres", Password = password, + Database = database, SslMode = SslMode.Disable, Pooling = false, Timeout = 3 +}.ConnectionString; +string MyConnection(int port, string database = "") => new MySqlConnectionStringBuilder +{ + Server = "127.0.0.1", Port = (uint)port, UserID = "root", Password = password, + Database = database, SslMode = MySqlSslMode.Disabled, Pooling = false, ConnectionTimeout = 3 +}.ConnectionString; +await using (var connection = new NpgsqlConnection(PgConnection("postgres"))) +{ + await connection.OpenAsync(); Console.WriteLine("Actual PostgreSQL=" + connection.PostgreSqlVersion); +} +foreach (var port in myPorts) +{ + await using var connection = new MySqlConnection(MyConnection(port)); await connection.OpenAsync(); + Console.WriteLine("Actual MySQL family=" + connection.ServerVersion); +} +if (args.Length > 3) return; +async Task PgSql(string database, string sql) +{ + await using var connection = new NpgsqlConnection(PgConnection(database)); await connection.OpenAsync(); + await using var command = connection.CreateCommand(); command.CommandText = sql; await command.ExecuteNonQueryAsync(); +} +ConnectionProfile Pg(string database) => new() +{ + Provider = DatabaseProviderKind.PostgreSql, Host = "127.0.0.1", Port = pgPort, + Username = "postgres", Password = password, Database = database, TlsMode = ConnectionTlsMode.Disabled +}; +await PgSql("postgres", "CREATE DATABASE seed;CREATE DATABASE mysql_back;CREATE DATABASE maria_back;"); +await PgSql("seed", """ + CREATE TABLE records(id bigint NOT NULL, number bigint NOT NULL, note text COLLATE "C", payload bytea, + CONSTRAINT records_pkey PRIMARY KEY(id,number)); + CREATE INDEX records_number ON records(number ASC NULLS FIRST); + CREATE TABLE empty_table(id bigint, note text COLLATE "C", payload bytea); + INSERT INTO records SELECT number,number%2,CASE WHEN number%3=0 THEN NULL ELSE '繁中😀 '||number END, + CASE WHEN number%2=0 THEN NULL ELSE decode('00ff','hex') END FROM generate_series(1,1000) number; + INSERT INTO records VALUES(-9223372036854775808,0,NULL,''::bytea),(9223372036854775807,0,'',NULL), + (-1,0,'a',''::bytea),(-2,0,'a ',NULL),(-3,0,'ABC',NULL),(-4,0,'abc',NULL),(-5,0,'é',NULL); + """); +for (var index = 0; index < myPorts.Length; index++) +{ + var port = myPorts[index]; + var major = index == 0 ? 8 : 11; + var collation = index == 0 ? "utf8mb4_0900_bin" : "utf8mb4_nopad_bin"; + await using (var connection = new MySqlConnection(MyConnection(port))) + { + await connection.OpenAsync(); await using var command = connection.CreateCommand(); + command.CommandText = "CREATE DATABASE target CHARACTER SET utf8mb4 COLLATE " + collation; + await command.ExecuteNonQueryAsync(); + } + var target = new ConnectionProfile + { + Provider = DatabaseProviderKind.MySql, Host = "127.0.0.1", Port = port, + Username = "root", Password = password, Database = "target", TlsMode = ConnectionTlsMode.Disabled + }; + var preview = await LogicalMigrationService.PreviewAsync(Pg("seed"), target, major); + Console.WriteLine("PG -> " + major + " preview=" + preview.CanStart + " " + string.Join(";", preview.Blockers)); + if (!preview.CanStart) throw new Exception("Preflight refused."); + var result = await LogicalMigrationService.RunAsync(Pg("seed"), target, major, preview, "target"); + Console.WriteLine(result.Outcome + " " + result.Report); + if (result.Outcome != LogicalMigrationOutcome.Verified) throw new Exception("Target migration failed."); + var back = Pg(index == 0 ? "mysql_back" : "maria_back"); + preview = await LogicalMigrationService.PreviewAsync(target, back, 16); + Console.WriteLine(major + " -> PG preview=" + preview.CanStart + " " + string.Join(";", preview.Blockers)); + if (!preview.CanStart) throw new Exception("Reverse preflight refused."); + result = await LogicalMigrationService.RunAsync(target, back, 16, preview, back.Database); + if (result.Outcome != LogicalMigrationOutcome.Verified) throw new Exception(result.Report); + await using (var connection = new MySqlConnection(MyConnection(port, "target"))) + { + await connection.OpenAsync(); await using var command = connection.CreateCommand(); + command.CommandText = "INSERT INTO records(id,number) VALUES(1,1)"; + try { await command.ExecuteNonQueryAsync(); throw new Exception("Duplicate primary key accepted."); } + catch (MySqlException exception) when (exception.Number == 1062) { } + } + Console.WriteLine("PASS actual PG16 <-> " + (index == 0 ? "MySQL8.4" : "MariaDB11.4") + "; 1007 rows + empty table, integer composite PK/index, UTF8/emoji/NULL/empty bytes/text/Int64 extrema, production schema/SHA256 verification"); + async Task MySql(string database, string sql) + { + await using var connection = new MySqlConnection(MyConnection(port, database)); await connection.OpenAsync(); + await using var command = connection.CreateCommand(); command.CommandText = sql; await command.ExecuteNonQueryAsync(); + } + async Task Count(string database, string sql) + { + await using var connection = new MySqlConnection(MyConnection(port, database)); await connection.OpenAsync(); + await using var command = connection.CreateCommand(); command.CommandText = sql; + return Convert.ToInt64(await command.ExecuteScalarAsync()); + } + async Task Empty(string database) + { + await MySql("", "CREATE DATABASE " + database + " CHARACTER SET utf8mb4 COLLATE " + collation); + var profile = target.Clone(); profile.Database = database; return profile; + } + var nonempty = await LogicalMigrationService.PreviewAsync(Pg("seed"), target, major); + if (nonempty.CanStart) throw new Exception("Existing target accepted."); + foreach (var scenario in new[] { "cancelled", "batch_cancel", "batch_fault", "verify_fault", "collision", "ack_lost", "disconnect" }) + { + var isolated = await Empty(scenario); + var approved = await LogicalMigrationService.PreviewAsync(Pg("seed"), isolated, major); + if (!approved.CanStart) throw new Exception("Fault fixture preflight failed."); + using var cancellation = new CancellationTokenSource(); + var fired = false; + result = await LogicalMigrationService.RunWithMySqlObserverAsync(Pg("seed"), isolated, major, approved, cancellation.Token, + (stage, connection, transaction) => + { + if (fired) return; + if (scenario == "cancelled" && stage == "staging") { fired = true; cancellation.Cancel(); } + if (scenario == "batch_cancel" && stage == "batch") { fired = true; cancellation.Cancel(); } + if (scenario == "batch_fault" && stage == "batch") { fired = true; throw new IOException("Synthetic batch failure."); } + if (scenario == "verify_fault" && stage == "verify") + { + fired = true; + using var lookup = connection.CreateCommand(); lookup.Transaction = transaction; + lookup.CommandText = "SELECT TABLE_NAME FROM information_schema.TABLES WHERE TABLE_SCHEMA=DATABASE() AND LEFT(TABLE_NAME,4)='_mp_' ORDER BY TABLE_NAME DESC LIMIT 1"; + var table = Convert.ToString(lookup.ExecuteScalar()); + using var damage = connection.CreateCommand(); damage.Transaction = transaction; + damage.CommandText = "DELETE FROM `" + table + "`"; damage.ExecuteNonQuery(); + } + if (scenario == "collision" && stage == "publish") + { + fired = true; using var command = connection.CreateCommand(); + command.CommandText = "CREATE TABLE records(id bigint) ENGINE=InnoDB"; command.ExecuteNonQuery(); + } + if (scenario == "ack_lost" && stage == "published") { fired = true; throw new IOException("Synthetic lost acknowledgement after real RENAME."); } + if (scenario == "disconnect" && stage == "published") { fired = true; connection.Close(); } + }); + var expected = scenario is "collision" or "ack_lost" or "disconnect" ? LogicalMigrationOutcome.Uncertain : LogicalMigrationOutcome.StagingRemains; + if (!fired || result.Outcome != expected || !result.RetryBlocked) throw new Exception("Fault outcome mismatch: " + scenario + " " + result.Report); + if (scenario != "ack_lost" && !result.Report.Contains("_mp_", StringComparison.Ordinal)) throw new Exception("Missing staging residue report."); + if (scenario is "ack_lost" or "disconnect") + { + if (await Count(scenario, "SELECT count(*) FROM records") != 1007 || + await Count(scenario, "SELECT count(*) FROM information_schema.TABLES WHERE TABLE_SCHEMA=DATABASE() AND LEFT(TABLE_NAME,4)='_mp_'") != 0) + throw new Exception("Real publish did not precede acknowledgement fault."); + } + else + { + var stagingCount = await Count(scenario, "SELECT count(*) FROM information_schema.TABLES WHERE TABLE_SCHEMA=DATABASE() AND LEFT(TABLE_NAME,4)='_mp_'"); + if (stagingCount != (scenario == "cancelled" ? 1 : 2)) throw new Exception("Staging residue count."); + if (await Count(scenario, "SELECT count(*) FROM information_schema.TABLES WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='empty_table'") != 0) + throw new Exception("Partial publish leaked first target table."); + if (scenario != "collision") + { + var stageNames = new List(); + await using (var connection = new MySqlConnection(MyConnection(port, scenario))) + { + await connection.OpenAsync(); await using var command = connection.CreateCommand(); + command.CommandText = "SELECT TABLE_NAME FROM information_schema.TABLES WHERE TABLE_SCHEMA=DATABASE() AND LEFT(TABLE_NAME,4)='_mp_'"; + await using var reader = await command.ExecuteReaderAsync(); + while (await reader.ReadAsync()) stageNames.Add(reader.GetString(0)); + } + foreach (var name in stageNames) + if (await Count(scenario, "SELECT count(*) FROM `" + name + "`") != 0) throw new Exception("Failed staging DML did not roll back."); + } + } + var retry = await LogicalMigrationService.PreviewAsync(Pg("seed"), isolated, major); + if (retry.CanStart) throw new Exception("Blind retry with residue/published tables accepted."); + Console.WriteLine("PASS real " + major + " " + scenario + " => " + result.Outcome + "; retry blocked; owned residue/publish state checked"); + } + var stale = await Empty("stale"); + var oldPreview = await LogicalMigrationService.PreviewAsync(Pg("seed"), stale, major); + await PgSql("seed", "UPDATE records SET note='changed_" + index + "' WHERE id=1;"); + result = await LogicalMigrationService.RunAsync(Pg("seed"), stale, major, oldPreview, "stale"); + if (result.Outcome != LogicalMigrationOutcome.Refused || await Count("stale", "SELECT count(*) FROM information_schema.TABLES WHERE TABLE_SCHEMA=DATABASE()") != 0) + throw new Exception("Stale snapshot wrote target."); + foreach (var (database, definition) in new[] + { + ("unsigned_source", "CREATE TABLE records(id bigint unsigned) ENGINE=InnoDB"), + ("default_source", "CREATE TABLE records(id bigint DEFAULT 1) ENGINE=InnoDB"), + ("prefix_source", "CREATE TABLE records(note longtext, KEY records_note(note(12))) ENGINE=InnoDB"), + ("trigger_source", "CREATE TABLE records(id bigint) ENGINE=InnoDB;CREATE TRIGGER audit_insert BEFORE INSERT ON records FOR EACH ROW SET NEW.id=1"), + ("procedure_source", "CREATE TABLE records(id bigint) ENGINE=InnoDB;CREATE PROCEDURE hidden_logic() SELECT 1"), + ("view_source", "CREATE TABLE records(id bigint) ENGINE=InnoDB;CREATE VIEW hidden_view AS SELECT id FROM records"), + ("pad_source", "CREATE TABLE records(note longtext COLLATE utf8mb4_bin) ENGINE=InnoDB") + }) + { + var bad = await Empty(database); await MySql(database, definition); + var rejected = await LogicalMigrationService.PreviewAsync(bad, Pg("mysql_back"), 16); + if (rejected.CanStart || !rejected.Blockers.Any(reason => reason.StartsWith("來源：", StringComparison.Ordinal))) + throw new Exception("Catalog feature silently accepted: " + database); + Console.WriteLine("PASS actual catalog refuses " + major + " " + database); + } + Console.WriteLine("PASS stale same-count source refused before DDL; existing targets refused; real fault/residue and unsupported catalog checks for " + major); + var faultDatabase = "source_fault_" + index; + await PgSql("postgres", "CREATE DATABASE " + faultDatabase); + var pgFault = Pg(faultDatabase); + preview = await LogicalMigrationService.PreviewAsync(target, pgFault, 16); + await MySql("target", "UPDATE records SET note='mysql_source_changed' WHERE id=1"); + result = await LogicalMigrationService.RunAsync(target, pgFault, 16, preview, faultDatabase); + if (result.Outcome != LogicalMigrationOutcome.RolledBack) throw new Exception("MySQL source stale snapshot not refused transactionally."); + preview = await LogicalMigrationService.PreviewAsync(target, pgFault, 16); + result = await LogicalMigrationService.RunWithObserverAsync(target, pgFault, 16, preview, faultDatabase, CancellationToken.None, + (stage, connection, transaction) => { if (stage == "batch") throw new IOException("Synthetic PostgreSQL target batch failure."); }); + if (result.Outcome != LogicalMigrationOutcome.RolledBack || result.RetryBlocked) throw new Exception("PG target rollback semantics changed."); + await using (var connection = new NpgsqlConnection(PgConnection(faultDatabase))) + { + await connection.OpenAsync(); await using var command = connection.CreateCommand(); + command.CommandText = "SELECT count(*) FROM pg_class c JOIN pg_namespace n ON n.oid=c.relnamespace WHERE n.nspname='public'"; + if (Convert.ToInt64(await command.ExecuteScalarAsync()) != 0) throw new Exception("PG target schema residue after rollback."); + } + var oversize = await Empty("oversize_source"); + await MySql("oversize_source", "CREATE TABLE records(payload longblob) ENGINE=InnoDB; INSERT INTO records VALUES(REPEAT('x',16777216))"); + try + { + await LogicalMigrationService.PreviewAsync(oversize, pgFault, 16); + throw new Exception("Oversize BLOB accepted."); + } + catch (InvalidOperationException exception) when (exception.Message.Contains("1 MiB", StringComparison.Ordinal)) { } + var preCancel = await Empty("pre_cancel"); + preview = await LogicalMigrationService.PreviewAsync(Pg("seed"), preCancel, major); + using (var cancellation = new CancellationTokenSource()) + { + cancellation.Cancel(); + try { await LogicalMigrationService.RunAsync(Pg("seed"), preCancel, major, preview, "pre_cancel", cancellation.Token); throw new Exception("Pre-cancel ignored."); } + catch (OperationCanceledException) { } + } + if (await Count("pre_cancel", "SELECT count(*) FROM information_schema.TABLES WHERE TABLE_SCHEMA=DATABASE()") != 0) + throw new Exception("Pre-cancel wrote objects."); + Console.WriteLine("PASS actual " + major + " source snapshot invalidation, PG partial write rollback with public objects=0, 16 MiB bounded BLOB refusal, pre-cancel empty target"); +} +await PgSql("postgres", "CREATE DATABASE custom_pk;CREATE DATABASE text_pk;CREATE DATABASE primary_named_source;CREATE DATABASE primary_named_target;"); +await PgSql("primary_named_source", "CREATE TABLE records(id bigint NOT NULL, CONSTRAINT \"PRIMARY\" PRIMARY KEY(id));INSERT INTO records VALUES(1);"); +var namedPreview = await LogicalMigrationService.PreviewAsync(Pg("primary_named_source"), Pg("primary_named_target"), 16); +if (!namedPreview.CanStart || !namedPreview.Statements.Any(sql => sql.Contains("CONSTRAINT \"PRIMARY\"", StringComparison.Ordinal))) + throw new Exception("PostgreSQL user-defined PRIMARY name misidentified as MySQL intrinsic name."); +var namedResult = await LogicalMigrationService.RunAsync(Pg("primary_named_source"), Pg("primary_named_target"), 16, namedPreview, "primary_named_target"); +if (namedResult.Outcome != LogicalMigrationOutcome.Verified) throw new Exception(namedResult.Report); +await PgSql("primary_named_target", "INSERT INTO records VALUES(1) ON CONFLICT ON CONSTRAINT \"PRIMARY\" DO NOTHING;"); +Console.WriteLine("PASS actual PG catalog preserves user-defined PRIMARY constraint, DDL/schema/SHA256 and ON CONFLICT; only MySQL intrinsic PRIMARY maps to PG default"); +await PgSql("custom_pk", "CREATE TABLE records(id bigint NOT NULL, CONSTRAINT explicit_records_pk PRIMARY KEY(id));"); +await PgSql("text_pk", "CREATE TABLE records(label text COLLATE \"C\" NOT NULL, CONSTRAINT records_pkey PRIMARY KEY(label));"); +foreach (var source in new[] { "custom_pk", "text_pk" }) +{ + await using var connection = new MySqlConnection(MyConnection(myPorts[0])); await connection.OpenAsync(); + await using var command = connection.CreateCommand(); command.CommandText = "CREATE DATABASE " + source + " CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_bin"; + await command.ExecuteNonQueryAsync(); + var target = new ConnectionProfile { Provider = DatabaseProviderKind.MySql, Host = "127.0.0.1", Port = myPorts[0], Username = "root", Password = password, Database = source, TlsMode = ConnectionTlsMode.Disabled }; + var denied = await LogicalMigrationService.PreviewAsync(Pg(source), target, 8); + if (denied.CanStart) throw new Exception("PG named/text key semantics lost silently."); + Console.WriteLine("PASS PG -> MySQL refuses " + source + " before DDL"); +} diff --git a/tests/MySqlLogicalMigrationHarness/Review.csproj b/tests/MySqlLogicalMigrationHarness/Review.csproj new file mode 100644 index 0000000..4255bd7 --- /dev/null +++ b/tests/MySqlLogicalMigrationHarness/Review.csproj @@ -0,0 +1,4 @@ + + Exenet8.0enableenableMySqlPunk.MySqlLogicalMigrationHarness + + diff --git a/tests/Run-MySqlLogicalMigration.py b/tests/Run-MySqlLogicalMigration.py new file mode 100644 index 0000000..fbea897 --- /dev/null +++ b/tests/Run-MySqlLogicalMigration.py @@ -0,0 +1,75 @@ +import hashlib +import json +import os +from pathlib import Path +import secrets +import subprocess +import sys +import time +import uuid + +root = Path(__file__).resolve().parents[1] +output = Path(sys.argv[1]) if len(sys.argv) > 1 else Path('/tmp') / ('mysqlpunk-my-pg-' + uuid.uuid4().hex) +output.mkdir(parents=True, exist_ok=False) +project = root / 'tests/MySqlLogicalMigrationHarness/Review.csproj' +subprocess.run(['dotnet', 'build', str(project), '-c', 'Release'], check=True) +environment = os.environ.copy() +password = 'Aa!' + secrets.token_hex(18) +environment.update(POSTGRES_PASSWORD=password, MYSQL_ROOT_PASSWORD=password, MARIADB_ROOT_PASSWORD=password, MYSQLPUNK_TEST_PASSWORD=password) +key = 'mysqlpunk-my-pg-' + uuid.uuid4().hex +metadata = {'images': [], 'sources': {}, 'cleanup': []} +containers = [] +for path in [project, *project.parent.glob('*.cs'), Path(__file__).resolve(), *sorted((root / 'mySQLPunk.Core').rglob('*.cs'))]: + if 'bin' not in path.parts and 'obj' not in path.parts: + metadata['sources'][str(path.relative_to(root))] = hashlib.sha256(path.read_bytes()).hexdigest() + +def docker(*arguments): + result = subprocess.run(['docker', *arguments], env=environment, text=True, capture_output=True, timeout=60) + if result.returncode: + raise RuntimeError('Docker operation failed: ' + arguments[0]) + return result.stdout.strip() + +try: + ports = [] + for index, (tag, port, directory, variable) in enumerate([ + ('postgres:16-alpine', 5432, '/var/lib/postgresql/data', 'POSTGRES_PASSWORD'), + ('mysql:8.4', 3306, '/var/lib/mysql', 'MYSQL_ROOT_PASSWORD'), + ('mariadb:11.4', 3306, '/var/lib/mysql', 'MARIADB_ROOT_PASSWORD') + ]): + image = json.loads(docker('image', 'inspect', tag, '--format', '{{json .}}')) + container = docker('run', '-d', '--name', key + '-' + str(index), '--label', 'mysqlpunk.test.owner=' + key, + '--cpus=1', '--memory=1g', '--pids-limit=512', '--mount', 'type=tmpfs,destination=' + directory + ',tmpfs-size=536870912', + '-e', variable, '-p', '127.0.0.1::' + str(port), image['Id']) + containers.append(container) + ports.append(docker('port', container, str(port) + '/tcp').split(':')[-1]) + metadata['images'].append({'tag': tag, 'id': image['Id'], 'digests': image.get('RepoDigests', [])}) + command = ['dotnet', str(project.parent / 'bin/Release/net8.0/MySqlPunk.MySqlLogicalMigrationHarness.dll'), *ports] + for attempt in range(60): + probe = subprocess.run(command + ['probe'], env=environment, text=True, capture_output=True, timeout=20) + if probe.returncode == 0: + metadata['versions'] = probe.stdout.strip().splitlines() + break + time.sleep(2) + else: + raise RuntimeError('Isolated engines not ready') + result = subprocess.run(command, env=environment, text=True, capture_output=True, timeout=240) + text = (result.stdout + '\n' + result.stderr).replace(password, '[redacted]') + (output / 'migration.log').write_text(text) + print(text, flush=True) + if result.returncode: + raise RuntimeError('Formal migration harness failed') +finally: + failures = [] + for container in containers: + try: + if docker('inspect', container, '--format', '{{index .Config.Labels "mysqlpunk.test.owner"}}') != key: + raise RuntimeError('Ownership mismatch; cleanup refused') + docker('rm', '-f', container) + metadata['cleanup'].append(container) + except Exception as exception: + failures.append({'container': container, 'error': str(exception)}) + metadata['cleanup_failures'] = failures + (output / 'evidence.json').write_text(json.dumps(metadata, indent=2) + '\n') + print('Evidence: ' + str(output), flush=True) + if failures: + raise RuntimeError('Fixture cleanup incomplete; inspect evidence.json')