AiDOPWarehouse/server/Plugins/Admin.NET.Plugin.AiDOP/DataPlatform/S0Dim/S0DimSqlBuilder.cs

using System.Text;

namespace Admin.NET.Plugin.AiDOP.DataPlatform.S0Dim;

/// <summary>
/// 由 <see cref="S0DimDefinition"/> 生成 SQL 的**唯一**位置（纯函数，不接触 DB，可单测）。
///
/// 生成的所有语句都必须能逐条证明租户范围：
/// staging 侧固定四段谓词 <c>tenant_id / source_system / source_table / sync_batch_id</c>；
/// dim 侧固定 <c>tenant_id</c>；source 侧固定 <c>tenant_id</c>。
/// </summary>
public static class S0DimSqlBuilder
{
    /// <summary>staging 别名。</summary>
    public const string StgAlias = "s";

    /// <summary>dim 别名。</summary>
    public const string DimAlias = "d";

    /// <summary>source 别名。</summary>
    public const string SrcAlias = "a";

    /// <summary>业务键各段之间的分隔符，与 <c>MdpStagingWriter.BuildBizKey</c> 的 <c>string.Join("#")</c> 一致。</summary>
    public const string BizKeySeparator = "#";

    /// <summary>checksum 行内字段分隔（US，业务值不可能包含）。</summary>
    private const string FieldSep = "CHAR(31 USING utf8mb4)";

    /// <summary>checksum 的 NULL 哨兵（RS），与空串可区分。</summary>
    private const string NullSentinel = "CHAR(30 USING utf8mb4)";

    private static string Q(string ident)
    {
        if (!S0DimDefinition.IdentifierRe.IsMatch(ident))
            throw new InvalidOperationException($"非法标识符：{ident}");
        return $"`{ident}`";
    }

    /// <summary>列取值表达式（作用于 staging 的 raw_data；TenantIdColumn 例外）。</summary>
    private static string ValueExpr(S0DimColumn c) => c.Kind switch
    {
        S0DimValueKind.TenantIdColumn => $"{StgAlias}.{Q(S0DimDefinition.TenantColumn)}",
        S0DimValueKind.Str => MdpJsonSql.Str(StgAlias, c.JsonPath!),
        S0DimValueKind.Int => MdpJsonSql.Int(StgAlias, c.JsonPath!),
        S0DimValueKind.Dec => MdpJsonSql.Dec(StgAlias, c.JsonPath!, c.Precision, c.Scale),
        S0DimValueKind.DateTimeSec => MdpJsonSql.DateTimeSec(StgAlias, c.JsonPath!),
        S0DimValueKind.BoolTrue => MdpJsonSql.BoolTrue(StgAlias, c.JsonPath!),
        _ => throw new InvalidOperationException($"未支持的取值方式：{c.Kind}")
    };

    /// <summary>
    /// staging 四段固定谓词。<paramref name="withBatch"/> 为 false 时用于「不限本批」的只读对账。
    /// </summary>
    public static string StagingFilter(bool withBatch = true)
    {
        var sb = new StringBuilder();
        sb.Append($"{StgAlias}.{Q(S0DimDefinition.TenantColumn)} = @TenantId");
        sb.Append($" AND {StgAlias}.`source_system` = @SourceSystem");
        sb.Append($" AND {StgAlias}.`source_table` = @SourceTable");
        if (withBatch) sb.Append($" AND {StgAlias}.`sync_batch_id` = @BatchId");
        return sb.ToString();
    }

    /// <summary>行级过滤：Required 列非空 + 布尔列可辨识（不满足的行不进 dim，随后被行数对账捕获）。</summary>
    private static IReadOnlyList<string> RowGuards(S0DimDefinition def)
    {
        var guards = new List<string>();
        foreach (var c in def.Columns)
        {
            if (c.Kind == S0DimValueKind.TenantIdColumn) continue;

            if (c.Required)
            {
                guards.Add($"({ValueExpr(c)}) IS NOT NULL");

                // IS NOT NULL 挡不住空串。业务键走 CONCAT_WS 拼接，空段会被**保留**，
                // 于是 Domain='' 会产出 '#LOC01' 这种「看起来合法」的键，而
                // AssertNoSourceDuplicateAsync 的 blank_cnt 只捕获整串全空的情况。
                // 当前 Catalog 中 Required 的 Str 列恰好就是全部业务键分量
                // （domain_code / work_center_code / department_code / location_code / shelf_code），
                // 故此守卫命中面 = 业务键，无附带影响。
                if (c.Kind == S0DimValueKind.Str)
                    guards.Add($"({ValueExpr(c)}) <> ''");
            }

            if (c.Kind == S0DimValueKind.BoolTrue)
            {
                // MdpJsonSql.BoolTrue 对 null / 不可识别值静默返回 0；此处强制要求原始值可辨识，
                // 否则整行被过滤 → dim_count < stg_count → 对账 FAIL（暴露而非吞掉）
                var raw = MdpJsonSql.Raw(StgAlias, c.JsonPath!);
                guards.Add($"LOWER({raw}) IN ('1','0','true','false')");
            }
        }
        return guards;
    }

    /// <summary>
    /// dim 物化语句：<c>INSERT INTO dim (...) SELECT ... FROM staging</c>。
    /// **刻意不带 ON DUPLICATE KEY UPDATE** —— 源侧业务键重复时直接撞 dim 唯一键报错，
    /// 由外层事务回滚（transform fail），绝不 last-wins。
    /// 参数：@TenantId @SourceSystem @SourceTable @BatchId @Now
    /// </summary>
    public static string BuildInsertSql(S0DimDefinition def)
    {
        def.Validate();

        var cols = new List<string>();
        var vals = new List<string>();
        foreach (var c in def.Columns)
        {
            cols.Add(Q(c.TargetColumn));
            vals.Add(ValueExpr(c));
        }

        cols.AddRange(["`source_system`", "`source_biz_key`", "`sync_batch_id`", "`sync_time`"]);
        vals.AddRange([$"{StgAlias}.`source_system`", $"{StgAlias}.`source_biz_key`", "@BatchId", "@Now"]);

        var where = new List<string> { StagingFilter() };
        where.AddRange(RowGuards(def));

        return $"""
                INSERT INTO {Q(def.DimTable)}
                  ({string.Join(", ", cols)})
                SELECT
                  {string.Join(",\n  ", vals)}
                FROM {Q(def.StagingTable)} {StgAlias}
                WHERE {string.Join("\n  AND ", where)}
                """;
    }

    /// <summary>staging 分区清理（purge）。参数：@TenantId @SourceSystem @SourceTable</summary>
    public static string BuildPurgeStagingSql(S0DimDefinition def) =>
        $"""
         DELETE {StgAlias} FROM {Q(def.StagingTable)} {StgAlias}
         WHERE {StagingFilter(withBatch: false)}
         """;

    /// <summary>源表行数（按租户）。参数：@TenantId</summary>
    public static string BuildSourceCountSql(S0DimDefinition def) =>
        $"SELECT COUNT(*) FROM {Q(def.SourceTable)} {SrcAlias} " +
        $"WHERE {SrcAlias}.{Q(S0DimDefinition.TenantColumn)} = @TenantId";

    /// <summary>源表中无法归属租户的行数（tenant_id IS NULL 或 &lt;= 0）—— 这些行整条链路都看不见。</summary>
    public static string BuildSourceUntenantedCountSql(S0DimDefinition def) =>
        $"SELECT COUNT(*) FROM {Q(def.SourceTable)} {SrcAlias} " +
        $"WHERE {SrcAlias}.{Q(S0DimDefinition.TenantColumn)} IS NULL " +
        $"OR {SrcAlias}.{Q(S0DimDefinition.TenantColumn)} <= 0";

    /// <summary>staging 行数。参数：@TenantId @SourceSystem @SourceTable [@BatchId]</summary>
    public static string BuildStagingCountSql(S0DimDefinition def, bool withBatch = true) =>
        $"SELECT COUNT(*) FROM {Q(def.StagingTable)} {StgAlias} WHERE {StagingFilter(withBatch)}";

    /// <summary>dim 行数（按租户）。参数：@TenantId</summary>
    public static string BuildDimCountSql(S0DimDefinition def) =>
        $"SELECT COUNT(*) FROM {Q(def.DimTable)} {DimAlias} " +
        $"WHERE {DimAlias}.{Q(S0DimDefinition.TenantColumn)} = @TenantId";

    /// <summary>dim 中不属于本批的残留行数（FULL Replace 之后必须为 0）。参数：@TenantId @BatchId</summary>
    public static string BuildDimBatchImpuritySql(S0DimDefinition def) =>
        $"SELECT COUNT(*) FROM {Q(def.DimTable)} {DimAlias} " +
        $"WHERE {DimAlias}.{Q(S0DimDefinition.TenantColumn)} = @TenantId " +
        $"AND {DimAlias}.`sync_batch_id` <> @BatchId";

    /// <summary>源侧业务键表达式（物理列，与 mdp_entity.biz_key_expr 同序同分隔）。</summary>
    public static string SourceBizKeyExpr(S0DimDefinition def) =>
        $"CONCAT_WS('{BizKeySeparator}', {string.Join(", ", def.SourceBizKeyColumns.Select(c => $"{SrcAlias}.{Q(c)}"))})";

    /// <summary>dim 侧业务键表达式（去掉 tenant_id 后的业务键列）。</summary>
    public static string DimBizKeyExpr(S0DimDefinition def) =>
        $"CONCAT_WS('{BizKeySeparator}', {string.Join(", ", def.BusinessKeyColumns.Skip(1).Select(c => $"{DimAlias}.{Q(c)}"))})";

    /// <summary>三层业务键集合来源子查询。</summary>
    public static string SourceBizKeySetSql(S0DimDefinition def) =>
        $"SELECT {SourceBizKeyExpr(def)} AS bk FROM {Q(def.SourceTable)} {SrcAlias} " +
        $"WHERE {SrcAlias}.{Q(S0DimDefinition.TenantColumn)} = @TenantId";

    /// <inheritdoc cref="SourceBizKeySetSql"/>
    public static string StagingBizKeySetSql(S0DimDefinition def, bool withBatch = true) =>
        $"SELECT {StgAlias}.`source_biz_key` AS bk FROM {Q(def.StagingTable)} {StgAlias} WHERE {StagingFilter(withBatch)}";

    /// <inheritdoc cref="SourceBizKeySetSql"/>
    public static string DimBizKeySetSql(S0DimDefinition def) =>
        $"SELECT {DimBizKeyExpr(def)} AS bk FROM {Q(def.DimTable)} {DimAlias} " +
        $"WHERE {DimAlias}.{Q(S0DimDefinition.TenantColumn)} = @TenantId";

    /// <summary>
    /// 集合差集计数：<paramref name="leftSql"/> \ <paramref name="rightSql"/>，
    /// 返回**左侧无匹配的行数**（不是去重后的键数）。
    ///
    /// <para>🔴 <b>不要写成 <c>LEFT JOIN … ON y.bk &lt;=&gt; x.bk WHERE y.bk IS NULL</c></b>（2026-09-07 实测）：
    /// 四个方向里总有一侧的 <c>bk</c> 是 <c>CONCAT_WS(...)</c> 计算表达式，**既不能建索引也不能哈希**。
    /// 当它落在 JOIN 右侧时，优化器只能退化成 nested loop —— 16150×16150 行的
    /// <c>stg \ src</c> 方向实测 <b>130.3s</b>（EXPLAIN: <c>rows=32,762,720</c>），
    /// 直接撞穿 <c>SqlSugarSetup.CommandTimeOut = 30</c>，而对称的 <c>src \ stg</c> 方向因右侧走
    /// <c>idx_stg_*_biz</c> 只要 0.5s —— <b>同一个缺陷只在一半方向上显形</b>，小表永远发现不了。
    /// 把 <c>&lt;=&gt;</c> 换成 <c>=</c> 无用（实测 127.7s），<c>NOT EXISTS</c> 也只到 69.2s。</para>
    ///
    /// <para>本写法把两侧各扫一遍后按 <c>bk</c> 聚合，复杂度 O(N+M)，与方向无关：
    /// 同一 <c>stg \ src</c> 实测 <b>0.5s</b>（260×），EXPLAIN 里 nested loop 消失。</para>
    ///
    /// <para>语义与旧写法逐条对齐：① 未匹配的左行在 LEFT JOIN 下恰好产出 1 行，
    /// 故 <c>SUM(l_cnt)</c> 与 <c>COUNT(*)</c> 等价，**左右两侧的重复行都不会改变结果**；
    /// ② <c>GROUP BY</c> 把所有 NULL 归为同一组，天然等价于 <c>&lt;=&gt;</c> 的 NULL 相等语义。</para>
    /// </summary>
    public static string BuildAntiJoinCountSql(string leftSql, string rightSql) =>
        $"""
         SELECT COALESCE(SUM(CASE WHEN g.r_cnt = 0 THEN g.l_cnt ELSE 0 END), 0) FROM (
           SELECT u.bk AS bk, SUM(u.l) AS l_cnt, SUM(u.r) AS r_cnt FROM (
             SELECT bk, 1 AS l, 0 AS r FROM ({leftSql}) la
             UNION ALL
             SELECT bk, 0 AS l, 1 AS r FROM ({rightSql}) ra
           ) u GROUP BY u.bk
         ) g
         """;

    /// <summary>集合内是否有重复 / 空业务键：返回 (总数, 去重数, 空值数)。</summary>
    public static string BuildBizKeyQualitySql(string setSql) =>
        $"""
         SELECT COUNT(*) AS total, COUNT(DISTINCT x.bk) AS distinct_cnt,
                SUM(CASE WHEN x.bk IS NULL OR x.bk = '' THEN 1 ELSE 0 END) AS blank_cnt
         FROM ({setSql}) x
         """;

    /// <summary>NULL 归一（checksum 用）。</summary>
    private static string Norm(string expr, S0DimValueKind kind) => kind switch
    {
        S0DimValueKind.DateTimeSec => $"IFNULL(DATE_FORMAT({expr}, '%Y-%m-%d %H:%i:%s'), {NullSentinel})",
        S0DimValueKind.BoolTrue => $"IFNULL(CAST(CAST({expr} AS UNSIGNED) AS CHAR), {NullSentinel})",
        _ => $"IFNULL(CAST({expr} AS CHAR), {NullSentinel})"
    };

    private static IReadOnlyList<S0DimColumn> ChecksumColumns(S0DimDefinition def) =>
        def.Columns.Where(c => c.Kind != S0DimValueKind.TenantIdColumn).ToList();

    /// <summary>
    /// 单行哈希：MD5 前 15 个十六进制位 → 60 bit 无符号整数。
    ///
    /// 🔴 <c>CAST(... AS UNSIGNED)</c> **不可省略**：<c>CONV()</c> 返回的是**字符串**，
    /// 而 MySQL 的 <c>SUM(字符串)</c> 会按 DOUBLE 累加 —— 只有约 16 位有效数字，
    /// 19 位的和末几位不可靠，且**两侧扫描顺序不同会产生不同舍入**，
    /// 于是完全相同的数据也会算出不同校验和（2026-09-07 Batch 2 实测：
    /// 同一租户 source=…980000 / dim=…980700，count 相同却误报不一致）。
    /// 转成整数后 <c>SUM</c> 走 DECIMAL 精确累加，与顺序无关。
    /// </summary>
    private static string RowHash(string rowStringExpr) =>
        $"CAST(CONV(SUBSTRING(MD5({rowStringExpr}),1,15),16,10) AS UNSIGNED)";

    /// <summary>
    /// 属性校验和：<c>SUM(<see cref="RowHash"/>)</c>。
    /// 用 SUM 而非 GROUP_CONCAT/BIT_XOR：交换律使**排序与结果无关**（无需 ORDER BY，也不受 group_concat_max_len 截断）；
    /// XOR 会让成对重复相互抵消从而掩盖重复。必须与 COUNT 成对使用。
    /// </summary>
    public static string BuildSourceChecksumSql(S0DimDefinition def)
    {
        var parts = ChecksumColumns(def).Select(c => Norm($"{SrcAlias}.{Q(c.JsonPath!)}", c.Kind));
        var rowStr = $"CONCAT_WS({FieldSep}, {Norm(SourceBizKeyExpr(def), S0DimValueKind.Str)}, {string.Join(", ", parts)})";
        return $"SELECT COUNT(*) AS cnt, CAST(IFNULL(SUM({RowHash(rowStr)}),0) AS DECIMAL(40,0)) AS chk " +
               $"FROM {Q(def.SourceTable)} {SrcAlias} WHERE {SrcAlias}.{Q(S0DimDefinition.TenantColumn)} = @TenantId";
    }

    /// <inheritdoc cref="BuildSourceChecksumSql"/>
    public static string BuildDimChecksumSql(S0DimDefinition def)
    {
        var parts = ChecksumColumns(def).Select(c => Norm($"{DimAlias}.{Q(c.TargetColumn)}", c.Kind));
        var rowStr = $"CONCAT_WS({FieldSep}, {Norm(DimBizKeyExpr(def), S0DimValueKind.Str)}, {string.Join(", ", parts)})";
        return $"SELECT COUNT(*) AS cnt, CAST(IFNULL(SUM({RowHash(rowStr)}),0) AS DECIMAL(40,0)) AS chk " +
               $"FROM {Q(def.DimTable)} {DimAlias} WHERE {DimAlias}.{Q(S0DimDefinition.TenantColumn)} = @TenantId";
    }

    /// <summary>dim 业务键重复计数（唯一键之外的兜底断言）。参数：@TenantId</summary>
    public static string BuildDimDuplicateBizKeySql(S0DimDefinition def) =>
        $"""
         SELECT COUNT(*) FROM (
           SELECT {DimBizKeyExpr(def)} AS bk
           FROM {Q(def.DimTable)} {DimAlias}
           WHERE {DimAlias}.{Q(S0DimDefinition.TenantColumn)} = @TenantId
           GROUP BY bk HAVING COUNT(*) > 1
         ) z
         """;

    /// <summary>镜像唯一性断言（如 dim_location 的 (tenant_id, location_code)）。参数：@TenantId</summary>
    public static string BuildMirrorUniqueViolationSql(S0DimDefinition def)
    {
        if (def.MirrorUniqueColumns is not { Count: > 0 }) throw new InvalidOperationException($"[{def.Key}] 未声明 MirrorUniqueColumns");
        var cols = string.Join(", ", def.MirrorUniqueColumns.Select(c => $"{DimAlias}.{Q(c)}"));
        return $"""
                SELECT COUNT(*) FROM (
                  SELECT {cols}
                  FROM {Q(def.DimTable)} {DimAlias}
                  WHERE {DimAlias}.{Q(S0DimDefinition.TenantColumn)} = @TenantId
                  GROUP BY {cols} HAVING COUNT(*) > 1
                ) z
                """;
    }

    /// <summary>孤儿子行（父维度中找不到父行）。参数：@TenantId</summary>
    /// <summary>
    /// 孤儿判定的公共 FROM/WHERE —— 样本查询与计数查询**共用同一份**，避免两处口径漂移。
    ///
    /// 「未建立关系」不是孤儿：父键列为 NULL 表示该行压根没声明父（如员工未分配部门），
    /// 与「声明了父但父不存在」（真悬挂）是两种语义。<c>&lt;=&gt;</c> 是 NULL 安全等值，
    /// 而父维度的业务键列均为 NOT NULL，故不加 <c>IS NOT NULL</c> 过滤会把全部未分配行
    /// 误报成孤儿，淹没真正的悬挂引用。对父键全部 Required 的维度（如 LocationShelf）本条件恒真，行为不变。
    /// </summary>
    private static string OrphanFromWhere(S0DimDefinition def)
    {
        var on = string.Join(" AND ", def.ParentKeyColumns!.Select(c => $"p.{Q(c)} <=> {DimAlias}.{Q(c)}"));
        var declared = string.Join(" AND ", def.ParentKeyColumns!.Select(c => $"{DimAlias}.{Q(c)} IS NOT NULL"));
        return $"""
                FROM {Q(def.DimTable)} {DimAlias}
                LEFT JOIN {Q(def.ParentDimTable!)} p
                  ON p.{Q(S0DimDefinition.TenantColumn)} = {DimAlias}.{Q(S0DimDefinition.TenantColumn)} AND {on}
                WHERE {DimAlias}.{Q(S0DimDefinition.TenantColumn)} = @TenantId AND {declared} AND p.`id` IS NULL
                """;
    }

    /// <summary>孤儿子行**样本**（受 limit 截断）。参数：@TenantId</summary>
    public static string BuildOrphanChildSql(S0DimDefinition def, int limit)
    {
        AssertParentDeclared(def);
        var sel = string.Join(", ", def.BusinessKeyColumns.Skip(1).Select(c => $"{DimAlias}.{Q(c)}"));
        return $"SELECT {sel} {OrphanFromWhere(def)} LIMIT {limit}";
    }

    /// <summary>孤儿子行**真实总数**（不截断）。参数：@TenantId</summary>
    public static string BuildOrphanChildCountSql(S0DimDefinition def)
    {
        AssertParentDeclared(def);
        return $"SELECT COUNT(*) {OrphanFromWhere(def)}";
    }

    private static void AssertParentDeclared(S0DimDefinition def)
    {
        if (def.ParentDimTable is null || def.ParentKeyColumns is not { Count: > 0 })
            throw new InvalidOperationException($"[{def.Key}] 未声明父维度");
    }

    // ════════════════════════════════════════════════════════════════════════════════════
    // 同库集合式贴源（source → staging）
    //
    // 背景：共享的 MdpStagingWriter 对**每一行**发一条 INSERT ... ON DUPLICATE KEY UPDATE
    // （MdpStagingWriter.cs 的 ExecuteCommandAsync），是为跨源（T8 SQLServer / dopdemorq /
    // Excel / API）设计的通用最小公倍数。当源与目标恰好是同一个物理库时，这些往返是纯浪费：
    // 实测 aidopdev 上约 65ms/行、稳定 ~15 行/s，ItemMaster 单租户 16150 行需 ≈20 分钟。
    //
    // 本段只**新增**一条同库快路径，不改 MdpStagingWriter —— 它被 187 个 entity 共用
    // （S0=20 / S1=34 / S2=16 / S3=44 / S4=8 / S5=29 / S6=15 / S7=12 / T8=9）。
    // 不满足同库判定时由调用方原样回退旧路径。
    // ════════════════════════════════════════════════════════════════════════════════════

    /// <summary>
    /// <c>source_row_id</c> 的候选列，**顺序必须与 <c>MdpDbPullExecutor.ResolveSourceRowId</c> 逐字一致**，
    /// 否则同一源行在两条路径下会得到不同的 <c>source_row_id</c>，贴源唯一键随之漂移。
    /// </summary>
    public static readonly IReadOnlyList<string> SourceRowIdCandidates =
    [
        "RecID", "RecId", "recid", "Recid",
        "id", "Id", "ID",
        "noid", "billno", "BillNo", "djbh", "Djbh"
    ];

    /// <summary>
    /// 快路径支持的 <c>DATA_TYPE</c>。**白名单而非黑名单**：出现任何未列出的类型
    /// （<c>binary</c> / <c>blob</c> / <c>json</c> / <c>enum</c> / <c>set</c> / <c>time</c> / <c>geometry</c> …）
    /// 一律判定不适用并回退旧路径，绝不猜测其序列化形态 ——
    /// 例如 <c>blob</c> 在旧路径是 <c>byte[]</c> → base64 字符串，而 <c>JSON_OBJECT</c> 会直出原始字节，
    /// 两者不等价且不会报错，属静默失真。
    /// （实测：S0 全部 19 张源表只用到 varchar/bigint/datetime/tinyint/decimal/int/bit 七类，全部在列。）
    /// </summary>
    public static readonly IReadOnlySet<string> SameDbSupportedDataTypes =
        new HashSet<string>(StringComparer.OrdinalIgnoreCase)
        {
            "tinyint", "smallint", "mediumint", "int", "integer", "bigint", "year",
            "decimal", "numeric", "float", "double",
            "datetime", "timestamp", "date",
            "char", "varchar", "tinytext", "text", "mediumtext", "longtext",
            "bit"
        };

    private static readonly IReadOnlySet<string> IntegerDataTypes =
        new HashSet<string>(StringComparer.OrdinalIgnoreCase)
        { "tinyint", "smallint", "mediumint", "int", "integer", "bigint", "year" };

    /// <summary>
    /// 单列在 <c>JSON_OBJECT</c> 中的表达式，产出的 <c>JSON_TYPE</c> 必须与旧
    /// <c>JsonSerializer.Serialize(dict, RawDataJsonOptions)</c> **逐类型一致**。
    /// 下表每一行都在 aidopdev 上用同一批真实行对拍过（旧 raw_data vs 本表达式）：
    ///
    /// <list type="table">
    /// <item><term><c>tinyint(1)</c>（恰好）</term><description>BOOLEAN <c>true</c>/<c>false</c></description></item>
    /// <item><term><c>tinyint(1) unsigned zerofill</c></term><description>INTEGER <c>0</c>（TreatTinyAsBoolean 不生效）</description></item>
    /// <item><term><c>bit(1)</c></term><description>INTEGER <c>0</c></description></item>
    /// <item><term><c>decimal(18,5)</c></term><description>DOUBLE <c>0.0</c>（**不是** DECIMAL <c>0.00000</c>）</description></item>
    /// <item><term><c>datetime</c></term><description>STRING <c>2023-12-18 13:57:30.000000</c>（**不是** JSON DATETIME）</description></item>
    /// <item><term>字符类</term><description>STRING 直出</description></item>
    /// </list>
    /// SQL NULL 在各分支下都产出 JSON <c>null</c>，与旧路径的 C# <c>null</c> 一致。
    /// </summary>
    private static string JsonValueExpr(S0DimSourceColumn c)
    {
        var col = $"{SrcAlias}.{Q(c.Name)}";

        if (string.Equals(c.ColumnType, "tinyint(1)", StringComparison.OrdinalIgnoreCase))
            return $"CAST(IF({col} IS NULL, NULL, IF({col} <> 0, 'true', 'false')) AS JSON)";

        if (string.Equals(c.DataType, "bit", StringComparison.OrdinalIgnoreCase))
            return $"CAST({col} AS UNSIGNED)";

        if (c.DataType is "decimal" or "numeric" or "float" or "double"
            || string.Equals(c.DataType, "decimal", StringComparison.OrdinalIgnoreCase)
            || string.Equals(c.DataType, "numeric", StringComparison.OrdinalIgnoreCase)
            || string.Equals(c.DataType, "float", StringComparison.OrdinalIgnoreCase)
            || string.Equals(c.DataType, "double", StringComparison.OrdinalIgnoreCase))
            return $"CAST({col} AS DOUBLE)";

        if (string.Equals(c.DataType, "datetime", StringComparison.OrdinalIgnoreCase)
            || string.Equals(c.DataType, "timestamp", StringComparison.OrdinalIgnoreCase)
            || string.Equals(c.DataType, "date", StringComparison.OrdinalIgnoreCase))
            return $"DATE_FORMAT({col}, '%Y-%m-%d %H:%i:%s.%f')";

        // 整数类与字符类都由 JSON_OBJECT 直出（INTEGER / STRING），与旧路径一致
        return col;
    }

    /// <summary>
    /// 单个候选列的「非空且非纯空白」取值，语义对齐旧路径的
    /// <c>string.IsNullOrWhiteSpace(v.ToString())</c> 跳过逻辑。
    /// 注意旧路径返回的是**未 TRIM 的原值**，故此处只用 TRIM 判空、取值仍取原值。
    /// </summary>
    private static string NonBlankChar(string column) =>
        $"IF({SrcAlias}.{Q(column)} IS NULL OR TRIM(CAST({SrcAlias}.{Q(column)} AS CHAR)) = '', " +
        $"NULL, CAST({SrcAlias}.{Q(column)} AS CHAR))";

    /// <summary>
    /// <c>source_row_id</c> 表达式：按候选顺序取第一个非空白值。
    /// 旧路径在全部候选都空白时兜底 <c>Guid.NewGuid()</c>；本表达式**不生成 <c>UUID()</c>** ——
    /// 调用方须先确保存在一个 NOT NULL 的整数候选列（见 <see cref="BuildSameDbStagingInsertSql"/> 的前置条件），
    /// 否则判定不适用、回退旧路径。这样既避免了在 <c>INSERT ... SELECT</c> 里引入非确定函数，
    /// 也保证两条路径对同一行给出同一个 <c>source_row_id</c>。
    /// </summary>
    private static string RowIdExpr(IReadOnlyList<string> presentCandidates) =>
        presentCandidates.Count == 1
            ? NonBlankChar(presentCandidates[0])
            : $"COALESCE({string.Join(", ", presentCandidates.Select(NonBlankChar))})";

    /// <summary>
    /// <c>source_biz_key</c> 表达式，语义对齐 <c>MdpStagingWriter.BuildBizKey</c>：
    /// **任一分量为 NULL 或空串 → 整个业务键作废，回落 <c>source_row_id</c>**。
    ///
    /// <para>🔴 这里不能直接写 <c>CONCAT_WS('#', …)</c> —— <c>CONCAT_WS</c> 会**跳过** NULL 分量，
    /// 于是 <c>(NULL, 'B')</c> 会得到 <c>'B'</c>，而旧路径给的是 <c>source_row_id</c>。
    /// 两条路径由此产出不同的 <c>source_biz_key</c>，且四向集合比对会在**下一次**刷新才炸，
    /// 极难定位。</para>
    /// </summary>
    private static string BizKeyExpr(S0DimDefinition def, string rowIdExpr)
    {
        var guard = string.Join(" AND ", def.SourceBizKeyColumns.Select(
            c => $"({SrcAlias}.{Q(c)} IS NOT NULL AND CAST({SrcAlias}.{Q(c)} AS CHAR) <> '')"));
        var parts = string.Join(", ", def.SourceBizKeyColumns.Select(
            c => $"CAST({SrcAlias}.{Q(c)} AS CHAR)"));
        return $"CASE WHEN {guard} THEN CONCAT_WS('{BizKeySeparator}', {parts}) ELSE {rowIdExpr} END";
    }

    /// <summary>
    /// 同库集合式贴源语句：<c>INSERT INTO staging (...) SELECT ... FROM source WHERE tenant_id = @TenantId</c>。
    /// 一条语句完成整租户装载，DB 往返 <b>O(1)</b>，取代旧路径的 O(N)。
    ///
    /// <para><b>租户硬门</b>：<c>WHERE</c> 结构性包含 <c>{SrcAlias}.tenant_id = @TenantId</c>，
    /// 源行无有效租户即不进结果集 —— 与 <c>RequireMatchingSourceTenant=true</c> 的 fail-closed 语义一致，
    /// **绝不拿 ctx 兜底、绝不落 0**。</para>
    ///
    /// <para><b>为什么保留 <c>ON DUPLICATE KEY UPDATE</c></b>：现有 FULL 契约确实保证
    /// <c>S0DimMaterializer</c> 在 purge（三段谓词，本租户+本源系统+本源表）之后才装载，
    /// 单纯 <c>INSERT</c> 本不会撞键；但旧 <c>MdpStagingWriter</c> 是 upsert，
    /// 源侧出现重复 <c>source_row_id</c> 时它 last-wins 而不报错。保留 upsert 才能让两条路径
    /// 在**异常数据**下也同构，不因换路径而改变 FULL 语义。</para>
    /// </summary>
    /// <param name="def">维度定义。</param>
    /// <param name="sourceColumns">源表全部列（决定 <c>raw_data</c> 的 JSON 形态）。</param>
    /// <param name="stagingColumns">staging 表实际存在的列（<c>factory_id</c> / <c>company_id</c> 按存在与否可选）。</param>
    /// <param name="rowIdCandidates">源表中实际存在的 <c>source_row_id</c> 候选列，按候选顺序。</param>
    public static string BuildSameDbStagingInsertSql(
        S0DimDefinition def,
        IReadOnlyList<S0DimSourceColumn> sourceColumns,
        IReadOnlySet<string> stagingColumns,
        IReadOnlyList<string> rowIdCandidates)
    {
        if (sourceColumns.Count == 0)
            throw new InvalidOperationException($"[{def.Key}] 源表无列，无法生成同库贴源语句");
        if (rowIdCandidates.Count == 0)
            throw new InvalidOperationException($"[{def.Key}] 源表无 source_row_id 候选列");

        var rowId = RowIdExpr(rowIdCandidates);

        var json = "JSON_OBJECT(" + string.Join(", ", sourceColumns.Select(
            c => $"'{c.Name}', {JsonValueExpr(c)}")) + ")";

        var cols = new List<string> { S0DimDefinition.TenantColumn };
        var vals = new List<string> { $"{SrcAlias}.{Q(S0DimDefinition.TenantColumn)}" };

        // 与旧 MdpStagingWriter 对齐：staging 有该列才写，且 factory 走 NormalizeFactoryId 口径
        // （>0 且 != tenant 才保留，否则 NULL）。ItemMaster 无 factory_id/FactoryId 列 →
        // 表达式恒 NULL，与旧路径实测的 16150/16150 全 NULL 一致。
        if (stagingColumns.Contains("factory_id"))
        {
            var f = SourceFactoryExpr(sourceColumns);
            cols.Add("factory_id");
            vals.Add(f is null
                ? "NULL"
                : $"IF({f} > 0 AND {f} <> {SrcAlias}.{Q(S0DimDefinition.TenantColumn)}, {f}, NULL)");
        }
        if (stagingColumns.Contains("company_id"))
        {
            var c = FindColumn(sourceColumns, "company_id");
            cols.Add("company_id");
            vals.Add(c is null ? "NULL" : $"{SrcAlias}.{Q(c)}");
        }

        cols.AddRange(["source_system", "source_table", "source_row_id", "source_biz_key",
                       "raw_data", "sync_batch_id", "sync_time", "process_status", "create_time"]);
        vals.AddRange(["@SourceSystem", "@SourceTable", rowId, BizKeyExpr(def, rowId),
                       json, "@BatchId", "@Now", "'PENDING'", "@Now"]);

        var updates = new List<string>
        {
            "`source_row_id`=VALUES(`source_row_id`)",
            "`raw_data`=VALUES(`raw_data`)",
            "`sync_batch_id`=VALUES(`sync_batch_id`)",
            "`sync_time`=VALUES(`sync_time`)",
            "`process_status`='PENDING'",
            $"`{S0DimDefinition.TenantColumn}`=VALUES(`{S0DimDefinition.TenantColumn}`)"
        };
        if (stagingColumns.Contains("factory_id")) updates.Add("`factory_id`=VALUES(`factory_id`)");
        if (stagingColumns.Contains("company_id")) updates.Add("`company_id`=VALUES(`company_id`)");
        updates.Add("`update_time`=@Now");

        return $"""
                INSERT INTO {Q(def.StagingTable)}
                  ({string.Join(", ", cols.Select(Q))})
                SELECT
                  {string.Join(",\n  ", vals)}
                FROM {Q(def.SourceTable)} {SrcAlias}
                WHERE {SrcAlias}.{Q(S0DimDefinition.TenantColumn)} = @TenantId
                ON DUPLICATE KEY UPDATE
                  {string.Join(",\n  ", updates)}
                """;
    }

    /// <summary>源侧工厂列：对齐旧路径的 <c>factory_id ?? FactoryId</c> 顺序；都不存在返回 null。</summary>
    private static string? SourceFactoryExpr(IReadOnlyList<S0DimSourceColumn> cols)
    {
        var hit = FindColumn(cols, "factory_id") ?? FindColumn(cols, "FactoryId");
        return hit is null ? null : $"{SrcAlias}.{Q(hit)}";
    }

    private static string? FindColumn(IReadOnlyList<S0DimSourceColumn> cols, string name) =>
        cols.FirstOrDefault(c => string.Equals(c.Name, name, StringComparison.OrdinalIgnoreCase))?.Name;
}