feat(s0): 支持同库集合式物化并落地物料维度

物料主数据 ItemMaster 是 S0 第一张大表（单租户 16150 行，此前 19 个维度
最大 292 行），把两个此前无从暴露的框架级缺陷同时打了出来：

一、source→staging 是 O(N) 远程往返
共享 MdpStagingWriter 对每一行发一条 upsert，实测 ~65ms/行、~15 行/s。
新增 S0DimSameDbStagingLoader：当源与本应用是同一物理库时，改用一条
INSERT ... SELECT 装载整租户。同库判定不解析主机串（localhost 与
127.0.0.1 会互相误判），改为两侧比对 @@server_uuid 与 DATABASE()；
判定只看源与库的客观属性，不看 entity 名、不看 source_code、不硬编码主机。
任一条件不成立即回退原逐行路径。

二、四向对账反连接是 O(N×M)，且只在一半方向显形
BuildAntiJoinCountSql 原用 LEFT JOIN ... ON y.bk <=> x.bk，而四个方向里
总有一侧的 bk 是 CONCAT_WS 计算表达式，既不能索引也不能哈希。它落在右侧时
退化为 nested loop：stg\src 实测 130.3s（rows=32,762,720），撞穿全局
CommandTimeOut=30；对称的 src\stg 因走索引仅 0.5s。改为两侧各扫一遍后按
bk 聚合，O(N+M)、与方向无关，同一查询 0.5s。

实测（租户 797，16150 行）：两处都未修 FAILED@1220s；只修反连接
SUCCESS@1169.8s；两处都修 SUCCESS@21.2s。两条路径产出的 dim 校验和逐位
相同，source_row_id / source_biz_key / tenant_id / factory_id 与源逐行
比对 16150/16150 一致。Com_insert_select 增量在 16150/5958/466 行下恒为 2。

raw_data 逐类型对齐旧 JsonSerializer 产出：tinyint(1) 走 JSON 布尔，而
tinyint(1) unsigned zerofill 与 bit(1) 走整数（TreatTinyAsBoolean 只对
恰好 tinyint(1) 生效，故须比对 COLUMN_TYPE 全串）；decimal 转 DOUBLE、
datetime 转 MySQL 字面字符串。类型处理用白名单，遇 blob/json/enum 等
一律整表回退，不猜其序列化形态。source_biz_key 复刻 BuildBizKey 的
「任一分量为空则整键作废、回落 source_row_id」语义，不能直接用 CONCAT_WS。

dim_item 身份为 tenant_id + item_code：ItemNum 跨租户碰撞 663 码 1534 行，
租户必须入身份；旧唯一键含 factory_ref_id，不予继承——它连 f(tenant) 都
不是，且 8010 被三个租户共用。RecID 仅作 legacy_source_id 技术属性供 C3
对账，不入身份、不得用代理 id 建外键。75 列按「填充率≥5% 且 distinct≥2」
取舍，排除影子列 expire_alarm_day、含 NULL 的 bit(1) SuppWarranty 与可空
tinyint DEAControlled。

共享入站核心 MdpStagingWriter / MdpDbPullExecutor / MdpSourceScopeFactory /
MdpJsonSql 零改动，S1-S9 与 T8 未触及。跨源大表仍走逐行路径，另案处理。

migration 保留 1.0.510（开发库已按本文件 SHA256 记账并 verify Success，
改名会让所有实例的 hash 守卫失配）；csproj 因远端 1.0.510 已被占用而取
1.0.511，两者不同号系版本竞争所致。

server/Admin.NET.Web.Entry/Admin.NET.Web.Entry.csproj

@@ -11,9 +11,9 @@
     <GenerateSatelliteAssembliesForCore>true</GenerateSatelliteAssembliesForCore>
     <Copyright>Admin.NET</Copyright>
     <Description>Admin.NET 通用权限开发平台</Description>
-    <AssemblyVersion>1.0.510</AssemblyVersion>
-    <FileVersion>1.0.510</FileVersion>
-    <Version>1.0.510</Version>
+    <AssemblyVersion>1.0.511</AssemblyVersion>
+    <FileVersion>1.0.511</FileVersion>
+    <Version>1.0.511</Version>
   </PropertyGroup>
 
   <ItemGroup>
@@ -739,6 +739,12 @@
     <None Update="UpdateScripts\1.0.509.verify.sql">
       <CopyToOutputDirectory>Always</CopyToOutputDirectory>
     </None>
+    <None Update="UpdateScripts\1.0.510.sql">
+      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
+    </None>
+    <None Update="UpdateScripts\1.0.510.verify.sql">
+      <CopyToOutputDirectory>Always</CopyToOutputDirectory>
+    </None>
     <None Update="UpdateScripts\UAT-PLACEHOLDER-MENU-HIDE.ops.sql">
       <CopyToOutputDirectory>Always</CopyToOutputDirectory>
     </None>

server/Admin.NET.Web.Entry/UpdateScripts/1.0.510.sql

@@ -0,0 +1,141 @@
+-- =====================================================================================
+-- 1.0.510  ·  S0 MASTER DATA PLATFORM · PHASE 2 · dim_item（大数据量门）
+--
+--   ItemMaster → mdp_stg_s0_item → dim_item
+--
+-- 【为什么是 dim_item 而不是复用 mdp_std_item —— 已封板，勿再讨论】
+--   mdp_std_item 是 Supply/S3 的**多源**物料目录：实测混入 ic_item（另一业务域主数据），
+--   租户 797 有 75 个 item_code 只来自 ic_item、在 ItemMaster 里 0 命中；其写入 SQL
+--   （S3MdpSyncTransformService）取值用 COALESCE($.ItemNum,$.number) 且不过滤 source_table/source_system。
+--   物理上它也无法当父维度：只有 6 个业务列，无 Domain、无 ItemMaster.RecID，
+--   两个最大租户的 factory_id/company_id 100% NULL。
+--   dim_item 不是第二份权威，是 ItemMaster 的第一次物化；二者消费方实测零重叠。
+--
+-- 【身份 = tenant_id + ItemNum】
+--   实测租户内重复 0、空白 0、非法租户 0；ItemNum 跨租户碰撞 663 码 / 1534 行 ⇒ 租户必须入身份。
+--   旧 UK uk_ItemMaster_factory_item 含 factory_ref_id，**不予继承**：它连 f(tenant) 都不是
+--   （租户 838257186181189 有 3 个取值含 0），且 8010 被三个租户共用。
+--
+-- 【大数据量】单租户最大 16150 行 > 默认 batch_size 5000，AssertNoSourceDuplicateAsync 会硬失败。
+--   本 entity 的 batch_size 设 50000（3.1x 头寸）。该值是 **entity 级**（取自 mdp_entity.batch_size），
+--   **不影响其它任何维度**，绝不是全局放大。
+--
+-- 【字段取舍】75 列 → 只物化「全库填充率 ≥5% 且全库 distinct ≥2」的列；
+--   并无条件排除：审计列 · 影子列 · bit(1) 列 · 可空 tinyint(1)。据此排除的典型：
+--     expire_alarm_day  0 填充的影子列（真列是 ExpireAlarmDays 22682）
+--     SuppWarranty      bit(1) 且含 2 行 NULL —— Int 会让对账每轮必挂、BoolTrue 会滤掉那 2 行
+--     DEAControlled     可空 tinyint(1)，2 行 NULL 会被 BoolTrue 守卫滤掉
+--     Length/Size/Height/Width/cust_supplied/enable_bom_version  全库 distinct=1，零信息
+--     Status            7/22684，且真实状态语义在 IsActive
+--     DefaultShelf/Remark/OwnerApplication/BOMDesignTime/RoutingDesignTime/
+--     drawing_design/design_owner/bom_design/routing_des/bom_version  全库 0 填充
+--     SizeUM(5)/LocationType(1)/PMBOM(797 空)                        近空
+--     CreateUser/CreateTime/UpdateUser                               审计列
+--
+-- 【legacy_source_id】← RecID，**技术属性**，供 C3 的 ProductStructureMaster.ComponentMaterialId
+--   这类 legacy 代理键做对账解析。它不是身份；且**绝不能用 dim 的代理 id 建外键** ——
+--   每轮 FULL REPLACE 后 dim_item.id 都会变。
+-- =====================================================================================
+
+CREATE TABLE IF NOT EXISTS `mdp_stg_s0_item` (
+  `id`              bigint       NOT NULL AUTO_INCREMENT,
+  `tenant_id`       bigint       NOT NULL DEFAULT 0,
+  `factory_id`      bigint       DEFAULT NULL,
+  `source_system`   varchar(50)  NOT NULL DEFAULT 'AIDOPDEV_MYSQL',
+  `source_table`    varchar(100) NOT NULL,
+  `source_row_id`   varchar(100) NOT NULL,
+  `source_biz_key`  varchar(300) DEFAULT NULL,
+  `sync_batch_id`   varchar(100) NOT NULL,
+  `sync_time`       datetime     NOT NULL,
+  `process_status`  varchar(20)  NOT NULL DEFAULT 'PENDING',
+  `raw_data`        json         NOT NULL,
+  `process_message` varchar(500) DEFAULT NULL,
+  `update_time`     datetime     DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+  `create_time`     datetime     DEFAULT CURRENT_TIMESTAMP,
+  PRIMARY KEY (`id`),
+  UNIQUE KEY `uk_stg_s0_item` (`tenant_id`,`source_system`,`source_table`,`source_row_id`),
+  KEY `idx_stg_s0_item_biz` (`source_biz_key`),
+  KEY `idx_stg_s0_item_batch` (`tenant_id`,`sync_batch_id`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='S0 贴源：物料（源 ItemMaster）';
+
+CREATE TABLE IF NOT EXISTS `dim_item` (
+  `id`                  bigint        NOT NULL AUTO_INCREMENT,
+  `tenant_id`           bigint        NOT NULL,
+  `item_code`           varchar(100)  NOT NULL COMMENT '物料编码（**唯一业务键**）← $.ItemNum',
+  `item_name`           varchar(200)  NOT NULL COMMENT '← $.Descr',
+  `item_name_2`         varchar(200)  DEFAULT NULL COMMENT '← $.Descr1',
+  `unit`                varchar(50)   DEFAULT NULL COMMENT '← $.UM',
+  `pur_mfg`             varchar(100)  DEFAULT NULL COMMENT '采购/制造属性 ← $.PurMfg',
+  `item_type`           varchar(100)  DEFAULT NULL COMMENT '物料类别 ← $.ItemType',
+  `commodity_code`      varchar(100)  DEFAULT NULL COMMENT '← $.CommodityCode',
+  `rev`                 varchar(50)   DEFAULT NULL COMMENT '← $.Rev',
+  `drawing`             varchar(100)  DEFAULT NULL COMMENT '← $.Drawing',
+  `pkg_code`            varchar(50)   DEFAULT NULL COMMENT '← $.PkgCode',
+  `series`              varchar(100)  DEFAULT NULL COMMENT '← $.Series',
+  `emt_type`            varchar(50)   DEFAULT NULL COMMENT '← $.EMTType',
+  `po_receipt_status`   varchar(8)    DEFAULT NULL COMMENT '← $.PORcptStatus',
+  `warranty_code`       varchar(255)  DEFAULT NULL COMMENT '← $.WarrantyCode',
+  `default_location`    varchar(100)  DEFAULT NULL COMMENT '← $.Location；软引用 dim_location，无 FK',
+  `buyer`               varchar(255)  DEFAULT NULL COMMENT '← $.Buyer',
+  `planner`             varchar(100)  DEFAULT NULL COMMENT '← $.Planner',
+  `net_weight`          decimal(18,5) DEFAULT NULL COMMENT '← $.NetWeight；精度必须与源逐位一致',
+  `net_weight_um`       varchar(50)   DEFAULT NULL COMMENT '← $.NetWeightUM',
+  `pur_lead_time`       int           DEFAULT NULL COMMENT '← $.PurLT',
+  `ins_lead_time`       int           DEFAULT NULL COMMENT '← $.InsLT',
+  `mfg_lead_time`       int           DEFAULT NULL COMMENT '← $.MfgLT',
+  `mfg_mttr`            int           DEFAULT NULL COMMENT '← $.MFGMTTR',
+  `days_between_pm`     int           DEFAULT NULL COMMENT '← $.DaysBetweenPM',
+  `expire_alarm_days`   int           DEFAULT NULL COMMENT '← $.ExpireAlarmDays（真列；同表 expire_alarm_day 是 0 填充影子列）',
+  `min_order_qty`       decimal(18,5) DEFAULT NULL COMMENT '← $.MinOrd',
+  `max_order_qty`       decimal(18,5) DEFAULT NULL COMMENT '← $.MaxOrd',
+  `min_order_qty_sales` decimal(18,5) DEFAULT NULL COMMENT '← $.MinOrdSales',
+  `order_multiple`      decimal(18,5) DEFAULT NULL COMMENT '← $.OrdMult',
+  `safety_stock`        decimal(18,5) DEFAULT NULL COMMENT '← $.SafetyStk',
+  `stock_turnover`      decimal(18,5) DEFAULT NULL COMMENT '← $.StockTurnOver',
+  `is_active`           tinyint(1)    NOT NULL COMMENT '← $.IsActive（真实有效状态，非 Status）',
+  `is_confirmed`        tinyint(1)    NOT NULL COMMENT '← $.IsConfirm',
+  `key_item`            tinyint(1)    NOT NULL COMMENT '← $.KeyItem',
+  `inspect_required`    tinyint(1)    NOT NULL COMMENT '← $.Inspect',
+  `issue_policy`        tinyint(1)    NOT NULL COMMENT '← $.IssuePolicy',
+  `install`             tinyint(1)    NOT NULL COMMENT '← $.Install',
+  `lot_serial_control`  tinyint(1)    NOT NULL COMMENT '← $.LotSerialControl',
+  `allocate_single_lot` tinyint(1)    NOT NULL COMMENT '← $.AllocateSingleLot',
+  `auto_lot_nums`       tinyint(1)    NOT NULL COMMENT '← $.AutoLotNums',
+  `trace_detail`        tinyint(1)    NOT NULL COMMENT '← $.TraceDetail',
+  `is_main_master`      tinyint(1)    NOT NULL COMMENT '← $.IsMainMas',
+  `mrp_required`        tinyint(1)    NOT NULL COMMENT '← $.MRPRequired',
+  `domain_code`         varchar(50)   DEFAULT NULL COMMENT 'LEGACY_SCOPE ← $.Domain；**不是身份**',
+  `company_id`          bigint        NOT NULL COMMENT 'LEGACY_SCOPE ← $.company_ref_id；**不是身份**',
+  `factory_id`          bigint        NOT NULL COMMENT 'LEGACY_SCOPE ← $.factory_ref_id；**不是身份**（连 f(tenant) 都不是）',
+  `legacy_source_id`    bigint        NOT NULL COMMENT '技术属性 ← $.RecID；供 C3 legacy 代理键对账。**不是身份，禁止用 dim.id 建外键**',
+  `source_system`       varchar(50)   NOT NULL,
+  `source_biz_key`      varchar(300)  NOT NULL,
+  `source_updated_at`   datetime      DEFAULT NULL COMMENT '← $.UpdateTime（源 22682/22684）',
+  `sync_batch_id`       varchar(64)   NOT NULL,
+  `sync_time`           datetime      NOT NULL,
+  `create_time`         datetime      NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  `update_time`         datetime      NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+  PRIMARY KEY (`id`),
+  -- 索引字节 = 8 + 100*4 = 408 < 3072
+  UNIQUE KEY `uk_dim_item` (`tenant_id`,`item_code`),
+  KEY `idx_dim_item_batch` (`sync_batch_id`),
+  -- 供 C3 的 legacy 代理键（如 ProductStructureMaster.ComponentMaterialId）反查
+  KEY `idx_dim_item_legacy` (`tenant_id`,`legacy_source_id`)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci COMMENT='S0 主数据维度：物料。源 ItemMaster，FULL REPLACE。**S0 的 Item 权威**，与 Supply/S3 的多源 mdp_std_item 语义不同。';
+
+SET @src_id = (SELECT `id` FROM `mdp_source` WHERE `source_code` = 'AIDOPDEV_MYSQL' LIMIT 1);
+
+INSERT INTO `mdp_entity`
+  (`tenant_id`,`source_id`,`entity_code`,`entity_name`,`entity_type`,
+   `source_table_name`,`target_table_name`,`sync_mode`,`incr_column`,`batch_size`,
+   `biz_key_expr`,`status`,`create_time`,`update_time`)
+SELECT 0, @src_id, 'S0_ITEM', 'S0 物料主数据入站', 'TABLE',
+       'ItemMaster', 'mdp_stg_s0_item', 'FULL', NULL, 50000,
+       'ItemNum', 1, NOW(), NOW()
+WHERE @src_id IS NOT NULL
+  AND NOT EXISTS (SELECT 1 FROM `mdp_entity` WHERE `entity_code` = 'S0_ITEM');
+
+-- ── 回滚（注释，非可执行）──
+-- DROP TABLE IF EXISTS `dim_item`;
+-- DROP TABLE IF EXISTS `mdp_stg_s0_item`;
+-- DELETE FROM `mdp_entity` WHERE `tenant_id` = 0 AND `entity_code` = 'S0_ITEM';

server/Admin.NET.Web.Entry/UpdateScripts/1.0.510.verify.sql

@@ -0,0 +1,66 @@
+-- 1.0.510.verify.sql
+-- 契约（CURRENT.md > F-11）：禁 SET @var · 每条返回 1/0 · 一条一断言 · 末条为真 SELECT。
+
+SELECT (SELECT COUNT(*) FROM information_schema.TABLES
+        WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME IN ('mdp_stg_s0_item','dim_item')) = 2;
+
+SELECT IFNULL((SELECT GROUP_CONCAT(COLUMN_NAME ORDER BY SEQ_IN_INDEX)
+               FROM information_schema.STATISTICS
+               WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'mdp_stg_s0_item'
+                 AND INDEX_NAME = 'uk_stg_s0_item' AND NON_UNIQUE = 0), '')
+       = 'tenant_id,source_system,source_table,source_row_id';
+
+SELECT IFNULL((SELECT GROUP_CONCAT(COLUMN_NAME ORDER BY SEQ_IN_INDEX)
+               FROM information_schema.STATISTICS
+               WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'dim_item'
+                 AND INDEX_NAME = 'uk_dim_item' AND NON_UNIQUE = 0), '')
+       = 'tenant_id,item_code';
+
+SELECT (SELECT COUNT(*) FROM information_schema.STATISTICS
+        WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'dim_item'
+          AND INDEX_NAME = 'uk_dim_item'
+          AND COLUMN_NAME IN ('domain_code','company_id','factory_id','legacy_source_id')) = 0;
+
+SELECT (SELECT COUNT(*) FROM information_schema.COLUMNS
+        WHERE TABLE_SCHEMA = DATABASE()
+          AND TABLE_NAME IN ('mdp_stg_s0_item','dim_item')
+          AND COLUMN_NAME = 'tenant_id' AND IS_NULLABLE = 'NO') = 2;
+
+SELECT (SELECT COUNT(*) FROM information_schema.COLUMNS
+        WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'dim_item'
+          AND COLUMN_NAME = 'legacy_source_id' AND IS_NULLABLE = 'NO') = 1;
+
+SELECT (SELECT COUNT(*) FROM information_schema.STATISTICS
+        WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'dim_item'
+          AND INDEX_NAME = 'idx_dim_item_legacy') = 2;
+
+SELECT (SELECT COUNT(*) FROM information_schema.COLUMNS
+        WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'dim_item'
+          AND COLUMN_NAME IN ('net_weight','min_order_qty','max_order_qty','min_order_qty_sales',
+                              'order_multiple','safety_stock','stock_turnover')
+          AND NUMERIC_PRECISION = 18 AND NUMERIC_SCALE = 5) = 7;
+
+SELECT (SELECT COUNT(*) FROM information_schema.COLUMNS
+        WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'dim_item'
+          AND COLUMN_NAME IN ('supp_warranty','dea_controlled','expire_alarm_day','item_status')) = 0;
+
+SELECT (SELECT COUNT(*) FROM information_schema.COLUMNS
+        WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = 'dim_item'
+          AND COLUMN_NAME = 'expire_alarm_days') = 1;
+
+SELECT (SELECT COUNT(*) FROM `mdp_entity`
+        WHERE `entity_code` = 'S0_ITEM'
+          AND `biz_key_expr` COLLATE utf8mb4_bin = 'ItemNum'
+          AND `source_table_name` COLLATE utf8mb4_bin = 'ItemMaster'
+          AND `target_table_name` COLLATE utf8mb4_bin = 'mdp_stg_s0_item'
+          AND `status` = 1 AND `sync_mode` = 'FULL' AND `incr_column` IS NULL
+          AND `batch_size` = 50000) = 1;
+
+SELECT (SELECT COUNT(*) FROM `mdp_entity`
+        WHERE `entity_code` LIKE 'S0\_%' AND `batch_size` <> 50000 AND `batch_size` <> 5000) = 0;
+
+SELECT (SELECT COUNT(*) FROM `mdp_entity`
+        WHERE `entity_code` LIKE 'S0\_%' AND `batch_size` = 50000) = 1;
+
+SELECT (SELECT COUNT(*) FROM `mdp_entity`
+        WHERE `source_table_name` = 'ItemMaster' AND `target_table_name` = 'mdp_stg_s0_item') = 1;

server/Plugins/Admin.NET.Plugin.AiDOP.Tests/S0/Dim/S0DimContractTests.cs

@@ -34,6 +34,8 @@ public class S0DimContractTests
                 "LABEL_TYPE", "SOP_FILE_TYPE", "QUALITY_BASE_TYPE",
                 "SAMPLING_SCHEME", "INSPECTION_ITEM", "INSPECTION_METHOD",
                 "INSPECTION_FREQUENCY", "INSPECTION_INSTRUMENT",
+                // C1 扩展
+                "ITEM",
                 // C3
                 "EMP_WORK_DUTY", "LINE_POST", "LINE_POST_SKILL"
             ],
@@ -698,6 +700,244 @@ public class S0DimContractTests
         }
     }
 
+    // ════════════════════════════════════════════════════════════════════════════
+    // 同库集合式贴源快路径（S0DimSameDbStagingLoader / BuildSameDbStagingInsertSql）
+    //
+    // 这些断言钉死的是「快路径与旧逐行 MdpStagingWriter 的语义等价」。每一条对应一个
+    // 在 aidopdev 上实测过的差异点；改动这些表达式前先看注释里的实测值。
+    // ════════════════════════════════════════════════════════════════════════════
+
+    /// <summary>构造一列源列描述，便于各用例复用。</summary>
+    private static S0DimSourceColumn Col(string name, string dataType, string? columnType = null, bool nullable = true)
+        => new(name, dataType, columnType ?? dataType, nullable);
+
+    private static readonly IReadOnlySet<string> StgCols =
+        new HashSet<string>(StringComparer.OrdinalIgnoreCase)
+        { "tenant_id", "factory_id", "source_system", "source_table", "source_row_id",
+          "source_biz_key", "raw_data", "sync_batch_id", "sync_time", "process_status", "create_time", "update_time" };
+
+    private static string SameDbSql(params S0DimSourceColumn[] cols) =>
+        S0DimSqlBuilder.BuildSameDbStagingInsertSql(
+            S0DimCatalog.Item, cols, StgCols, ["RecID"]);
+
+    [Fact]
+    public void SameDb_unsigned_integer_json_type_label_is_downstream_irrelevant()
+    {
+        // 已知差异（如实登记，非缺陷）：bit(1) 与 `tinyint(1) unsigned zerofill` 在
+        // 旧路径的 raw_data 里 JSON_TYPE = INTEGER，快路径则是 UNSIGNED INTEGER。
+        //
+        // 为什么不去消除它：downstream 只经 MdpJsonSql 读值，而它一律走
+        // JSON_UNQUOTE(JSON_EXTRACT(...)) 再 CAST —— 两种标签下取出的文本完全相同。
+        // 实证：同一批 16150 行真实数据，旧路径与快路径产出的 dim 校验和**逐位相同**
+        // （9215906568012255809101），即该标签差异对 dim 契约零影响。
+        //
+        // 本断言钉死「MdpJsonSql 不依赖 JSON_TYPE」这一前提 —— 一旦有人改成按类型分支，
+        // 上述等价性即不再成立，此测试会失败并把人引到这段注释。
+        foreach (var expr in new[]
+                 {
+                     MdpJsonSql.Str("s", "X"), MdpJsonSql.Int("s", "X"),
+                     MdpJsonSql.Dec("s", "X", 18, 5), MdpJsonSql.DateTimeSec("s", "X"),
+                     MdpJsonSql.BoolTrue("s", "X"), MdpJsonSql.Raw("s", "X")
+                 })
+            Assert.DoesNotContain("JSON_TYPE", expr, Ord);
+    }
+
+    [Fact]
+    public void SameDb_source_column_must_be_sqlsugar_mappable()
+    {
+        // 🔴 SqlSugar 的 IDataReaderEntityBuilder.CreateBuilder 走反射构造，
+        // 对只有主构造函数的 positional record 会抛 ArgumentNullException (Parameter 'con')。
+        // 该异常会被 S0DimSameDbStagingLoader 吞掉并静默回退逐行路径 ——
+        // **表现为「变慢」而不是「报错」**，实测踩过一次。此断言让回归变成编译期/单测期可见。
+        var t = typeof(S0DimSourceColumn);
+        Assert.NotNull(t.GetConstructor(Type.EmptyTypes));
+        foreach (var prop in new[] { "Name", "DataType", "ColumnType", "IsNullable" })
+        {
+            var p = t.GetProperty(prop);
+            Assert.NotNull(p);
+            Assert.True(p!.CanWrite, $"{prop} 必须可写，否则 SqlSugar 无法回填");
+        }
+    }
+
+    [Fact]
+    public void SameDb_tinyint1_must_emit_json_boolean_not_integer()
+    {
+        // 旧路径：MySqlConnector 的 TreatTinyAsBoolean 把 tinyint(1) 变成 CLR bool
+        // → JsonSerializer 产出 JSON BOOLEAN（实测 mdp_stg_s0_item.$.IsActive = BOOLEAN "true"）。
+        // JSON_OBJECT 直出会是 INTEGER 1 —— 必须用 CAST(... AS JSON) 纠正。
+        var sql = SameDbSql(Col("RecID", "bigint", nullable: false), Col("IsActive", "tinyint", "tinyint(1)"));
+        Assert.Contains("CAST(IF(a.`IsActive` IS NULL, NULL, IF(a.`IsActive` <> 0, 'true', 'false')) AS JSON)", sql, Ord);
+    }
+
+    [Fact]
+    public void SameDb_zerofill_tinyint1_must_stay_integer()
+    {
+        // 🔴 TreatTinyAsBoolean 对 `tinyint(1) unsigned zerofill` **不生效**
+        // （实测 $.enable_bom_version = INTEGER 0，而 $.IsActive = BOOLEAN）。
+        // 判定必须比对 COLUMN_TYPE 全串而非 DATA_TYPE，否则这一列会被错误地转成布尔。
+        var sql = SameDbSql(Col("RecID", "bigint", nullable: false),
+                            Col("enable_bom_version", "tinyint", "tinyint(1) unsigned zerofill"));
+        Assert.DoesNotContain("`enable_bom_version` IS NULL, NULL, IF(", sql, Ord);
+        Assert.Contains("'enable_bom_version', a.`enable_bom_version`", sql, Ord);
+    }
+
+    [Fact]
+    public void SameDb_bit_must_cast_unsigned_never_char()
+    {
+        // bit(1) 在旧路径是 INTEGER（实测 $.SuppWarranty = INTEGER 0）。
+        // 若直出，JSON_OBJECT 会得到裸字节；CAST AS CHAR 更糟（0x01 而非 '1'，见 S0DimValueKind.Int 注释）。
+        var sql = SameDbSql(Col("RecID", "bigint", nullable: false), Col("SuppWarranty", "bit", "bit(1)"));
+        Assert.Contains("'SuppWarranty', CAST(a.`SuppWarranty` AS UNSIGNED)", sql, Ord);
+    }
+
+    [Fact]
+    public void SameDb_decimal_must_emit_double_not_decimal()
+    {
+        // 旧路径 decimal → CLR decimal → JSON DOUBLE «0.0»；JSON_OBJECT 直出是 DECIMAL «0.00000»。
+        // JSON_UNQUOTE 后字符串不同 → staging 侧任何按原文比对都会差异。
+        var sql = SameDbSql(Col("RecID", "bigint", nullable: false), Col("NetWeight", "decimal", "decimal(18,5)"));
+        Assert.Contains("'NetWeight', CAST(a.`NetWeight` AS DOUBLE)", sql, Ord);
+    }
+
+    [Fact]
+    public void SameDb_datetime_must_emit_mysql_literal_string()
+    {
+        // 旧路径用 MysqlLiteralDateTimeConverter 写成 STRING "yyyy-MM-dd HH:mm:ss.ffffff"
+        // （实测 $.UpdateTime = STRING "2023-12-18 13:57:30.000000"）；JSON_OBJECT 直出是 JSON DATETIME。
+        var sql = SameDbSql(Col("RecID", "bigint", nullable: false), Col("UpdateTime", "datetime"));
+        Assert.Contains("'UpdateTime', DATE_FORMAT(a.`UpdateTime`, '%Y-%m-%d %H:%i:%s.%f')", sql, Ord);
+    }
+
+    [Fact]
+    public void SameDb_string_columns_must_pass_through_untouched()
+    {
+        // 空串必须保持空串（不能被当成 NULL）；字符类直出即 JSON STRING，与旧路径一致。
+        var sql = SameDbSql(Col("RecID", "bigint", nullable: false), Col("Descr", "varchar", "varchar(200)"));
+        Assert.Contains("'Descr', a.`Descr`", sql, Ord);
+    }
+
+    [Fact]
+    public void SameDb_biz_key_must_fall_back_to_row_id_when_any_part_is_null_or_blank()
+    {
+        // 🔴 MdpStagingWriter.BuildBizKey：任一分量 null/空串 → 整个 biz key 作废 → 回落 source_row_id。
+        // 直接写 CONCAT_WS 会**跳过** NULL 分量而产出「看起来合法」的短键，与旧路径不等价，
+        // 且要到下一轮四向集合比对才炸。
+        var sql = SameDbSql(Col("RecID", "bigint", nullable: false), Col("ItemNum", "varchar", "varchar(100)"));
+        Assert.Contains("CASE WHEN (a.`ItemNum` IS NOT NULL AND CAST(a.`ItemNum` AS CHAR) <> '')", sql, Ord);
+        Assert.Contains("ELSE IF(a.`RecID` IS NULL OR TRIM(CAST(a.`RecID` AS CHAR)) = '', NULL, CAST(a.`RecID` AS CHAR)) END", sql, Ord);
+    }
+
+    [Fact]
+    public void SameDb_row_id_candidate_order_must_match_shared_pull_executor()
+    {
+        // 顺序与 MdpDbPullExecutor.ResolveSourceRowId 逐字一致；错序会让同一源行在两条路径下
+        // 得到不同 source_row_id，贴源唯一键随之漂移。
+        Assert.Equal(
+            ["RecID", "RecId", "recid", "Recid", "id", "Id", "ID", "noid", "billno", "BillNo", "djbh", "Djbh"],
+            S0DimSqlBuilder.SourceRowIdCandidates);
+    }
+
+    [Fact]
+    public void SameDb_sql_must_never_contain_nondeterministic_uuid()
+    {
+        // 旧路径在全部候选都空白时兜底 Guid.NewGuid()。快路径不复刻该分支，
+        // 而是由 Loader 要求「首个候选是 NOT NULL 整数」后判定适用，从源头排除。
+        var sql = SameDbSql(Col("RecID", "bigint", nullable: false), Col("ItemNum", "varchar", "varchar(100)"));
+        Assert.DoesNotContain("UUID(", sql, StringComparison.OrdinalIgnoreCase);
+    }
+
+    [Fact]
+    public void SameDb_sql_must_carry_structural_tenant_gate()
+    {
+        // 租户硬门必须结构性存在，且绝不允许出现 ctx 兜底常量
+        var sql = SameDbSql(Col("RecID", "bigint", nullable: false), Col("ItemNum", "varchar", "varchar(100)"));
+        Assert.Contains("WHERE a.`tenant_id` = @TenantId", sql, Ord);
+        Assert.Contains("`tenant_id`", sql, Ord);
+        Assert.DoesNotContain("tenant_id = 0", sql, Ord);
+        Assert.DoesNotContain("IFNULL(a.`tenant_id`", sql, Ord);
+    }
+
+    [Fact]
+    public void SameDb_supported_types_must_be_an_allowlist_excluding_binary_and_json()
+    {
+        // 白名单而非黑名单：未知类型一律回退旧路径，不猜其序列化形态。
+        // blob 在旧路径是 byte[]→base64，JSON_OBJECT 会直出原始字节，静默失真。
+        foreach (var ok in new[] { "varchar", "bigint", "datetime", "tinyint", "decimal", "int", "bit" })
+            Assert.Contains(ok, S0DimSqlBuilder.SameDbSupportedDataTypes);
+        foreach (var no in new[] { "blob", "varbinary", "binary", "json", "enum", "set", "time", "geometry" })
+            Assert.DoesNotContain(no, S0DimSqlBuilder.SameDbSupportedDataTypes);
+    }
+
+    [Fact]
+    public void SameDb_must_preserve_full_semantics_with_upsert()
+    {
+        // 旧 MdpStagingWriter 是 upsert；保留 ON DUPLICATE KEY UPDATE 才能在源侧出现重复
+        // source_row_id 这类异常数据下两条路径同构，而不是一条报错一条 last-wins。
+        var sql = SameDbSql(Col("RecID", "bigint", nullable: false), Col("ItemNum", "varchar", "varchar(100)"));
+        Assert.Contains("ON DUPLICATE KEY UPDATE", sql, Ord);
+        Assert.Contains("`process_status`='PENDING'", sql, Ord);
+    }
+
+    [Fact]
+    public void AntiJoin_must_not_use_nested_loop_left_join_shape()
+    {
+        // 🔴 LEFT JOIN ... ON y.bk <=> x.bk 在「右侧是 CONCAT_WS 计算列」的方向上退化成 nested loop：
+        // 16150 行的 stg\src 实测 130.3s（rows=32,762,720），撞穿 CommandTimeOut=30；
+        // 而对称的 src\stg 只要 0.5s —— 缺陷只在一半方向显形，小表永远发现不了。
+        var sql = S0DimSqlBuilder.BuildAntiJoinCountSql("SELECT 1 AS bk", "SELECT 2 AS bk");
+        Assert.DoesNotContain("LEFT JOIN", sql, Ord);
+        Assert.Contains("UNION ALL", sql, Ord);
+        Assert.Contains("GROUP BY u.bk", sql, Ord);
+        // 必须按左行计数（SUM(l_cnt)）而非去重键数，才与旧 COUNT(*) 语义一致
+        Assert.Contains("SUM(CASE WHEN g.r_cnt = 0 THEN g.l_cnt ELSE 0 END)", sql, Ord);
+    }
+
+    [Fact]
+    public void Item_identity_must_be_tenant_and_item_code_only()
+    {
+        // 旧 UK uk_ItemMaster_factory_item 含 factory，不予继承：factory_ref_id 连 f(tenant) 都不是
+        // （838257186181189 有 3 个取值含 0），且 8010 被三个租户共用。
+        var d = S0DimCatalog.Item;
+        Assert.Equal(["tenant_id", "item_code"], d.BusinessKeyColumns);
+        Assert.Equal(["ItemNum"], d.SourceBizKeyColumns);
+        foreach (var legacy in new[] { "domain_code", "company_id", "factory_id", "legacy_source_id" })
+            Assert.DoesNotContain(legacy, d.BusinessKeyColumns);
+        Assert.Null(d.ParentDimTable);
+    }
+
+    [Fact]
+    public void Item_legacy_source_id_is_technical_attribute_only()
+    {
+        // 供 C3 的 ComponentMaterialId 这类 legacy 代理键做对账解析；绝不是身份。
+        // 也绝不能用 dim 的代理 id 建外键 —— 每轮 FULL REPLACE 后 dim_item.id 都会变。
+        var col = S0DimCatalog.Item.Columns.Single(c => c.TargetColumn == "legacy_source_id");
+        Assert.Equal("RecID", col.JsonPath);
+        Assert.Equal(S0DimValueKind.Int, col.Kind);
+    }
+
+    [Fact]
+    public void Item_must_not_materialize_bit_or_nullable_bool_or_shadow_columns()
+    {
+        // SuppWarranty 是 bit(1) 且含 2 行 NULL —— Int 会让对账必挂、BoolTrue 会滤掉那 2 行。
+        // DEAControlled 是可空 tinyint(1)，同样会被 BoolTrue 守卫滤行。
+        // expire_alarm_day 是 0 填充的影子列，真列为 ExpireAlarmDays。
+        var paths = S0DimCatalog.Item.Columns.Select(c => c.JsonPath ?? "").ToList();
+        foreach (var forbidden in new[] { "SuppWarranty", "DEAControlled", "expire_alarm_day" })
+            Assert.DoesNotContain(forbidden, paths);
+        Assert.Contains("ExpireAlarmDays", paths);
+    }
+
+    [Fact]
+    public void Item_must_precede_c3_definitions_in_catalog_order()
+    {
+        // 后续 6 张 C3 都要挂 dim_item 作父，父必须先就绪
+        var keys = S0DimCatalog.All.Select(d => d.Key).ToList();
+        var item = keys.IndexOf("ITEM");
+        Assert.True(item >= 0);
+        foreach (var child in new[] { "EMP_WORK_DUTY", "LINE_POST", "LINE_POST_SKILL" })
+            Assert.True(item < keys.IndexOf(child), $"ITEM 必须排在 {child} 之前");
+    }
+
     [Fact]
     public void Batch4_must_not_reuse_dopdemorq_staging_tables()
     {

server/Plugins/Admin.NET.Plugin.AiDOP/DataPlatform/S0Dim/S0DimCatalog.cs

@@ -612,6 +612,107 @@ public static class S0DimCatalog
             new("source_updated_at",      S0DimValueKind.DateTimeSec, "FMODIFYTIME")
         ]);
 
+    /// <summary>
+    /// 物料。源 <c>ItemMaster</c> —— **S0 的 Item 权威**（2026-09-07 拍板）。
+    ///
+    /// <para><b>与 <c>mdp_std_item</c> 的关系已封板，不再讨论</b>：那是 Supply/S3 的**多源**物料目录
+    /// （实测混入 <c>ic_item</c>，797 租户有 75 个 item_code 只来自 ic_item、在 ItemMaster 里 0 命中），
+    /// 且物理上无 <c>Domain</c>、无 <c>ItemMaster.RecID</c>、两个最大租户的 factory/company 100% NULL，
+    /// 无法承担 C3 的父维度职责。<c>dim_item</c> 不是第二份权威，是 <c>ItemMaster</c> 的第一次物化，
+    /// 二者语义域不相交（消费方实测零重叠）。</para>
+    ///
+    /// <para><b>身份 = tenant_id + ItemNum</b>。实测：租户内重复 <b>0</b>、空白 <b>0</b>、非法租户 <b>0</b>；
+    /// <c>ItemNum</c> 跨租户碰撞 <b>663 码 / 1534 行</b> ⇒ 租户必须入身份。
+    /// 旧唯一键 <c>uk_ItemMaster_factory_item</c> 含 <c>factory_ref_id</c>，**不予继承** ——
+    /// 它连 f(tenant) 都不是（租户 838257186181189 有 3 个取值，含 <c>0</c>），
+    /// 且 <c>8010</c> 被三个租户共用，放进唯一键既无判别力又把身份绑到跨租户共享值上。</para>
+    ///
+    /// <para>⚠️ <b>单租户最大 16150 行，超过默认 batch_size 5000</b> ——
+    /// <c>AssertNoSourceDuplicateAsync</c> 会硬失败。本 entity 的 <c>mdp_entity.batch_size</c> 设 <b>50000</b>。
+    /// 该值是 **entity 级**（取自 <c>entity.BatchSize</c>），不影响其它任何维度。</para>
+    ///
+    /// <para><c>legacy_source_id ← RecID</c> 是**技术属性**，供 C3 的
+    /// <c>ProductStructureMaster.ComponentMaterialId</c> 这类 legacy 代理键做对账解析。
+    /// **它绝不是身份，也不得用 dim 的代理 id 建立外键** —— 每轮 FULL REPLACE 后 <c>dim_item.id</c> 都会变。</para>
+    ///
+    /// <para>字段取舍规则（75 列 → 只物化符合以下全部条件的）：**全库填充率 ≥5% 且全库 distinct ≥2**；
+    /// 并无条件排除审计列、影子列、<c>bit(1)</c> 列、可空 <c>tinyint(1)</c>。据此排除的典型：
+    /// <c>expire_alarm_day</c>（0 填充的影子列，真列是 <c>ExpireAlarmDays</c> 22682）·
+    /// <c>SuppWarranty</c>（<c>bit(1)</c> 且含 2 行 NULL —— 用 Int 对账必挂、用 BoolTrue 会滤掉那 2 行）·
+    /// <c>DEAControlled</c>（可空 tinyint，2 行 NULL 同样会被 BoolTrue 守卫滤掉）·
+    /// <c>Length</c>/<c>Size</c>/<c>Height</c>/<c>Width</c>/<c>cust_supplied</c>/<c>enable_bom_version</c>
+    /// （全库 distinct=1，零信息）· <c>Status</c>（7/22684，且真实状态是 <c>IsActive</c>）。</para>
+    /// </summary>
+    public static readonly S0DimDefinition Item = new(
+        Key: "ITEM",
+        EntityCode: "S0_ITEM",
+        SourceTable: "ItemMaster",
+        SourceSystem: LocalSourceSystem,
+        StagingTable: "mdp_stg_s0_item",
+        DimTable: "dim_item",
+        BusinessKeyColumns: ["tenant_id", "item_code"],
+        SourceBizKeyColumns: ["ItemNum"],
+        Columns:
+        [
+            new("tenant_id",           S0DimValueKind.TenantIdColumn),
+            // ── CORE
+            new("item_code",           S0DimValueKind.Str,         "ItemNum", Required: true),
+            new("item_name",           S0DimValueKind.Str,         "Descr",   Required: true),
+            // ── ATTRIBUTE · 描述与分类
+            new("item_name_2",         S0DimValueKind.Str,         "Descr1"),
+            new("unit",                S0DimValueKind.Str,         "UM"),
+            new("pur_mfg",             S0DimValueKind.Str,         "PurMfg"),
+            new("item_type",           S0DimValueKind.Str,         "ItemType"),
+            new("commodity_code",      S0DimValueKind.Str,         "CommodityCode"),
+            new("rev",                 S0DimValueKind.Str,         "Rev"),
+            new("drawing",             S0DimValueKind.Str,         "Drawing"),
+            new("pkg_code",            S0DimValueKind.Str,         "PkgCode"),
+            new("series",              S0DimValueKind.Str,         "Series"),
+            new("emt_type",            S0DimValueKind.Str,         "EMTType"),
+            new("po_receipt_status",   S0DimValueKind.Str,         "PORcptStatus"),
+            new("warranty_code",       S0DimValueKind.Str,         "WarrantyCode"),
+            // ── ATTRIBUTE · 库位与人员
+            new("default_location",    S0DimValueKind.Str,         "Location"),
+            new("buyer",               S0DimValueKind.Str,         "Buyer"),
+            new("planner",             S0DimValueKind.Str,         "Planner"),
+            // ── ATTRIBUTE · 计量
+            new("net_weight",          S0DimValueKind.Dec,         "NetWeight",   Precision: 18, Scale: 5),
+            new("net_weight_um",       S0DimValueKind.Str,         "NetWeightUM"),
+            // ── ATTRIBUTE · 计划参数（Dec 精度必须与源逐位一致，否则 CAST AS CHAR 后校验和 FAIL）
+            new("pur_lead_time",       S0DimValueKind.Int,         "PurLT"),
+            new("ins_lead_time",       S0DimValueKind.Int,         "InsLT"),
+            new("mfg_lead_time",       S0DimValueKind.Int,         "MfgLT"),
+            new("mfg_mttr",            S0DimValueKind.Int,         "MFGMTTR"),
+            new("days_between_pm",     S0DimValueKind.Int,         "DaysBetweenPM"),
+            new("expire_alarm_days",   S0DimValueKind.Int,         "ExpireAlarmDays"),
+            new("min_order_qty",       S0DimValueKind.Dec,         "MinOrd",        Precision: 18, Scale: 5),
+            new("max_order_qty",       S0DimValueKind.Dec,         "MaxOrd",        Precision: 18, Scale: 5),
+            new("min_order_qty_sales", S0DimValueKind.Dec,         "MinOrdSales",   Precision: 18, Scale: 5),
+            new("order_multiple",      S0DimValueKind.Dec,         "OrdMult",       Precision: 18, Scale: 5),
+            new("safety_stock",        S0DimValueKind.Dec,         "SafetyStk",     Precision: 18, Scale: 5),
+            new("stock_turnover",      S0DimValueKind.Dec,         "StockTurnOver", Precision: 18, Scale: 5),
+            // ── ATTRIBUTE · 标志位（全部是 tinyint(1) NOT NULL，可安全走 BoolTrue）
+            new("is_active",           S0DimValueKind.BoolTrue,    "IsActive"),
+            new("is_confirmed",        S0DimValueKind.BoolTrue,    "IsConfirm"),
+            new("key_item",            S0DimValueKind.BoolTrue,    "KeyItem"),
+            new("inspect_required",    S0DimValueKind.BoolTrue,    "Inspect"),
+            new("issue_policy",        S0DimValueKind.BoolTrue,    "IssuePolicy"),
+            new("install",             S0DimValueKind.BoolTrue,    "Install"),
+            new("lot_serial_control",  S0DimValueKind.BoolTrue,    "LotSerialControl"),
+            new("allocate_single_lot", S0DimValueKind.BoolTrue,    "AllocateSingleLot"),
+            new("auto_lot_nums",       S0DimValueKind.BoolTrue,    "AutoLotNums"),
+            new("trace_detail",        S0DimValueKind.BoolTrue,    "TraceDetail"),
+            new("is_main_master",      S0DimValueKind.BoolTrue,    "IsMainMas"),
+            new("mrp_required",        S0DimValueKind.BoolTrue,    "MRPRequired"),
+            // ── LEGACY_SCOPE（属性透传，绝不入唯一键）
+            new("domain_code",         S0DimValueKind.Str,         "Domain"),
+            new("company_id",          S0DimValueKind.Int,         "company_ref_id", Required: true),
+            new("factory_id",          S0DimValueKind.Int,         "factory_ref_id", Required: true),
+            // ── 技术属性：供 C3 的 legacy 代理键对账用（见类型注释）
+            new("legacy_source_id",    S0DimValueKind.Int,         "RecID",          Required: true),
+            new("source_updated_at",   S0DimValueKind.DateTimeSec, "UpdateTime")
+        ]);
+
     // ════════════════════════════════════════════════════════════════════════════
     // C3 · 关系 / 明细桥接（bridge_*）
     // ════════════════════════════════════════════════════════════════════════════
@@ -768,6 +869,8 @@ public static class S0DimCatalog
         LabelType, SopFileType, QualityBaseType,
         // C2-2 qms 字典族（共用 mdp_stg_s0_qms_dict）
         SamplingScheme, InspectionItem, InspectionMethod, InspectionFrequency, InspectionInstrument,
+        // C1 扩展：物料（S0 Item 权威；后续 6 张 C3 都要挂它作父，故必须排在 C3 之前）
+        Item,
         // C3 关系桥（EmpWorkDuty 的父 dim_employee 在前；LinePostSkill 的父 LinePost 紧邻其前）
         EmpWorkDuty, LinePost, LinePostSkill
     ];

server/Plugins/Admin.NET.Plugin.AiDOP/DataPlatform/S0Dim/S0DimMaterializer.cs

@@ -29,6 +29,7 @@ public sealed class S0DimMaterializer : ITransient
     private readonly ISqlSugarClient _db;
     private readonly MdpSourcePullDispatcher _pullDispatcher;
     private readonly S0DimStagingPurge _purge;
+    private readonly S0DimSameDbStagingLoader _sameDbLoader;
     private readonly S0DimReconciler _reconciler;
     private readonly ILogger<S0DimMaterializer> _logger;
 
@@ -37,12 +38,14 @@ public sealed class S0DimMaterializer : ITransient
         ISqlSugarClient db,
         MdpSourcePullDispatcher pullDispatcher,
         S0DimStagingPurge purge,
+        S0DimSameDbStagingLoader sameDbLoader,
         S0DimReconciler reconciler,
         ILogger<S0DimMaterializer> logger)
     {
         _db = db;
         _pullDispatcher = pullDispatcher;
         _purge = purge;
+        _sameDbLoader = sameDbLoader;
         _reconciler = reconciler;
         _logger = logger;
     }
@@ -85,9 +88,27 @@ public sealed class S0DimMaterializer : ITransient
                 FullRefresh = true,
                 RequireMatchingSourceTenant = true
             };
-            var pull = await _pullDispatcher.PullAllByEntityCodeAsync(def.EntityCode, pullCtx, ct);
-            result.SourcePulled = pull.RowsPulled;
-            result.StagingWritten = pull.RowsWritten;
+            //     同库时优先走集合式装载（一条 INSERT ... SELECT，DB 往返 O(1)）；
+            //     判定不成立或计划生成失败一律回退下面的逐行路径，语义不变、只是慢。
+            //     参见 S0DimSameDbStagingLoader：判定只依据源与库的客观属性，
+            //     不看 entity 名 / source_code 字面量 / 主机硬编码。
+            var samedb = await _sameDbLoader.TryPlanAsync(def, ct);
+            if (samedb is not null)
+            {
+                var written = await _sameDbLoader.LoadAsync(def, samedb, tenantId, batchId, ct);
+                // 集合式语句无「读了多少行」的独立计数，装载行数即贴源行数；
+                // 真正的把关在下面 [4][5] 的 stg == source 闸门，不依赖这里的自报数。
+                result.SourcePulled = written;
+                result.StagingWritten = written;
+                _logger.LogInformation("[S0Dim] {Key} 走同库快路径：{Reason}，写入 {Rows} 行",
+                    def.Key, samedb.Reason, written);
+            }
+            else
+            {
+                var pull = await _pullDispatcher.PullAllByEntityCodeAsync(def.EntityCode, pullCtx, ct);
+                result.SourcePulled = pull.RowsPulled;
+                result.StagingWritten = pull.RowsWritten;
+            }
 
             // [4][5] 闸门：本批 staging 必须与源侧行数完全一致，否则不碰 dim
             var stagingCount = await _db.Ado.GetIntAsync(S0DimSqlBuilder.BuildStagingCountSql(def), ps);

server/Plugins/Admin.NET.Plugin.AiDOP/DataPlatform/S0Dim/S0DimSameDbStagingLoader.cs

@@ -0,0 +1,157 @@
+using Admin.NET.Plugin.AiDOP.Entity.DataPlatform;
+using Admin.NET.Plugin.AiDOP.DataPlatform.Executors;
+using Microsoft.Extensions.Logging;
+using SqlSugar;
+
+namespace Admin.NET.Plugin.AiDOP.DataPlatform.S0Dim;
+
+/// <summary>
+/// 同库集合式贴源装载器：当维度的源与本应用目标库是**同一个物理库**时，
+/// 用一条 <c>INSERT ... SELECT</c> 完成整租户 source → staging 装载，DB 往返 O(1)。
+///
+/// <para><b>为什么需要它</b>：共享的 <c>MdpStagingWriter</c> 逐行发 upsert（为跨源设计的通用做法），
+/// 在 aidopdev 上实测约 65ms/行、稳定 ~15 行/s。前 19 个 S0 维度最大仅 292 行（秒级）故从未暴露；
+/// <c>ItemMaster</c> 单租户 16150 行需 ≈20 分钟，实测一次完整刷新 1220s 后仍未进 dim。</para>
+///
+/// <para><b>边界</b>：本类**只新增快路径，不修改 <c>MdpStagingWriter</c> / <c>MdpDbPullExecutor</c>**
+/// —— 它们被 187 个 entity 共用（S0/S1/S2/S3/S4/S5/S6/S7/T8）。
+/// 任何一项判定不成立即返回 <c>null</c>，由调用方原样回退旧路径。
+/// 判定**只依据源与库的客观属性**，不看 entity 名、不看 source_code 字面量、不硬编码主机。</para>
+/// </summary>
+public sealed class S0DimSameDbStagingLoader : ITransient
+{
+    private readonly ISqlSugarClient _db;
+    private readonly MdpSourceScopeFactory _scopeFactory;
+    private readonly ILogger<S0DimSameDbStagingLoader> _logger;
+
+    /// <summary>注入本库客户端与源连接工厂。</summary>
+    public S0DimSameDbStagingLoader(
+        ISqlSugarClient db, MdpSourceScopeFactory scopeFactory, ILogger<S0DimSameDbStagingLoader> logger)
+    {
+        _db = db;
+        _scopeFactory = scopeFactory;
+        _logger = logger;
+    }
+
+    /// <summary>同库装载计划；<c>null</c> 表示不适用。</summary>
+    /// <param name="Sql">已生成的 <c>INSERT ... SELECT</c> 语句。</param>
+    /// <param name="Reason">适用理由（写日志用）。</param>
+    public sealed record Plan(string Sql, string Reason);
+
+    /// <summary>
+    /// 判定是否可走同库快路径，可以则连同 SQL 一起返回；否则返回 <c>null</c> 并记录原因。
+    /// **任何异常都吞掉并返回 <c>null</c>** —— 快路径是纯优化，绝不能因判定本身失败而阻断刷新。
+    /// </summary>
+    public async Task<Plan?> TryPlanAsync(S0DimDefinition def, CancellationToken ct = default)
+    {
+        try
+        {
+            var source = await _db.Queryable<MdpSource>()
+                .Where(x => x.SourceCode == def.SourceSystem).FirstAsync(ct);
+
+            // ① 源必须是 DB 型 MySQL。非 DB（API / FILE_EXCEL）根本没有「同库」概念；
+            //    非 MySQL 无法在本库的一条语句里直接 SELECT。
+            if (source is null) return Decline(def, "mdp_source 未登记");
+            if (!string.Equals(source.SourceType, "DB", StringComparison.OrdinalIgnoreCase))
+                return Decline(def, $"source_type={source.SourceType} 非 DB");
+            if (!string.Equals(source.DbType, "MySQL", StringComparison.OrdinalIgnoreCase))
+                return Decline(def, $"db_type={source.DbType} 非 MySQL");
+
+            // ② 同库判定：不解析主机字符串（localhost / 127.0.0.1 / DNS 名会互相误判），
+            //    改为两侧各问一次数据库自己的身份。server_uuid 全局唯一、DATABASE() 精确到库。
+            var here = await IdentifyAsync(_db);
+            var there = await IdentifyAsync(await _scopeFactory.GetScopeAsync(source.SourceCode, ct));
+            if (here is null || there is null || !string.Equals(here, there, StringComparison.Ordinal))
+                return Decline(def, $"非同一物理库（本库={here ?? "?"} 源={there ?? "?"}）");
+
+            // ③ 源表必须有 tenant_id —— 租户硬门要靠它，没有就只能走旧路径的逐行判定
+            var srcCols = await ColumnsAsync(def.SourceTable);
+            if (srcCols.Count == 0) return Decline(def, "源表不存在或无列");
+            if (!srcCols.Any(c => string.Equals(c.Name, S0DimDefinition.TenantColumn, StringComparison.OrdinalIgnoreCase)))
+                return Decline(def, $"源表无 {S0DimDefinition.TenantColumn} 列");
+
+            // ④ 类型白名单：任何未支持类型都不猜其序列化形态，整表回退（见 SameDbSupportedDataTypes 注释）
+            var unsupported = srcCols.Where(c => !S0DimSqlBuilder.SameDbSupportedDataTypes.Contains(c.DataType))
+                                     .Select(c => $"{c.Name}:{c.DataType}").ToList();
+            if (unsupported.Count > 0)
+                return Decline(def, $"存在未支持类型 {string.Join(",", unsupported.Take(3))}");
+
+            // ⑤ source_row_id：必须存在候选列，且首个候选是 NOT NULL 整数
+            //    —— 这样它既不可能为 NULL 也不可能是空白，旧路径的 Guid 兜底分支永不触发，
+            //    快路径便无需在 INSERT ... SELECT 里引入非确定的 UUID()。
+            var present = S0DimSqlBuilder.SourceRowIdCandidates
+                .Select(cand => srcCols.FirstOrDefault(c => string.Equals(c.Name, cand, StringComparison.OrdinalIgnoreCase)))
+                .Where(c => c is not null).Select(c => c!).ToList();
+            if (present.Count == 0) return Decline(def, "源表无 source_row_id 候选列（会退化为 GUID）");
+            var first = present[0];
+            if (first.IsNullable || !S0DimSqlBuilder.SameDbSupportedDataTypes.Contains(first.DataType)
+                || !IsIntegerType(first.DataType))
+                return Decline(def, $"首个 row_id 候选 {first.Name} 非 NOT NULL 整数，无法排除 GUID 兜底");
+
+            // ⑥ 业务键列必须都在源表里
+            var missing = def.SourceBizKeyColumns
+                .Where(b => !srcCols.Any(c => string.Equals(c.Name, b, StringComparison.OrdinalIgnoreCase))).ToList();
+            if (missing.Count > 0) return Decline(def, $"源表缺业务键列 {string.Join(",", missing)}");
+
+            var stgCols = (await ColumnsAsync(def.StagingTable)).Select(c => c.Name).ToHashSet(StringComparer.OrdinalIgnoreCase);
+            if (stgCols.Count == 0) return Decline(def, "staging 表不存在");
+
+            var sql = S0DimSqlBuilder.BuildSameDbStagingInsertSql(
+                def, srcCols, stgCols, present.Select(c => c.Name).ToList());
+            return new Plan(sql, $"同库集合式（{srcCols.Count} 列 → {def.StagingTable}）");
+        }
+        catch (OperationCanceledException) { throw; }
+        catch (Exception ex)
+        {
+            _logger.LogWarning(ex, "[S0Dim] {Key} 同库快路径判定异常，回退旧路径", def.Key);
+            return null;
+        }
+    }
+
+    /// <summary>执行同库装载，返回受影响行数。</summary>
+    public async Task<int> LoadAsync(S0DimDefinition def, Plan plan, long tenantId, string batchId, CancellationToken ct = default)
+    {
+        // 与 S0DimStagingPurge 同款前置：绝不允许无效租户进到写路径
+        if (tenantId <= 0) throw new InvalidOperationException($"[{def.Key}] 同库装载拒绝无效租户：{tenantId}");
+
+        return await _db.Ado.ExecuteCommandAsync(plan.Sql, new List<SugarParameter>
+        {
+            new("@TenantId", tenantId),
+            new("@SourceSystem", def.SourceSystem),
+            new("@SourceTable", def.SourceTable),
+            new("@BatchId", batchId),
+            new("@Now", DateTime.Now)
+        });
+    }
+
+    private Plan? Decline(S0DimDefinition def, string reason)
+    {
+        _logger.LogInformation("[S0Dim] {Key} 不走同库快路径：{Reason}（回退逐行贴源）", def.Key, reason);
+        return null;
+    }
+
+    /// <summary>库身份指纹：<c>server_uuid</c> + 当前库名。两侧相同即同一物理库同一 schema。</summary>
+    private static async Task<string?> IdentifyAsync(ISqlSugarClient db)
+    {
+        var rows = await db.Ado.SqlQueryAsync<string>("SELECT CONCAT(@@server_uuid, '/', DATABASE())");
+        return rows?.FirstOrDefault();
+    }
+
+    private async Task<List<S0DimSourceColumn>> ColumnsAsync(string table) =>
+        await _db.Ado.SqlQueryAsync<S0DimSourceColumn>(
+            """
+            SELECT COLUMN_NAME AS Name, DATA_TYPE AS DataType, COLUMN_TYPE AS ColumnType,
+                   (IS_NULLABLE = 'YES') AS IsNullable
+            FROM information_schema.COLUMNS
+            WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = @t
+            ORDER BY ORDINAL_POSITION
+            """,
+            new SugarParameter("@t", table)) ?? [];
+
+    private static bool IsIntegerType(string dataType) =>
+        IntegerDataTypesForRowId.Contains(dataType);
+
+    private static readonly IReadOnlySet<string> IntegerDataTypesForRowId =
+        new HashSet<string>(StringComparer.OrdinalIgnoreCase)
+        { "tinyint", "smallint", "mediumint", "int", "integer", "bigint" };
+}

server/Plugins/Admin.NET.Plugin.AiDOP/DataPlatform/S0Dim/S0DimSourceColumn.cs

@@ -0,0 +1,44 @@
+namespace Admin.NET.Plugin.AiDOP.DataPlatform.S0Dim;
+
+/// <summary>
+/// 源表的一列在 <c>information_schema.COLUMNS</c> 里的形态，供
+/// <see cref="S0DimSqlBuilder.BuildSameDbStagingInsertSql"/> 决定该列的 JSON 表达式。
+///
+/// <para>之所以既要 <see cref="DataType"/> 又要 <see cref="ColumnType"/>：
+/// MySqlConnector 的 <c>TreatTinyAsBoolean</c>（默认开）只对**恰好** <c>tinyint(1)</c> 生效，
+/// 而 <c>tinyint(1) unsigned zerofill</c> 走整数路径 —— 两者 <c>DATA_TYPE</c> 同为 <c>tinyint</c>，
+/// 只有 <c>COLUMN_TYPE</c> 能区分。这条差异已在 aidopdev 上用既有 <c>raw_data</c> 实测证实
+/// （<c>IsActive</c>→BOOLEAN，<c>enable_bom_version</c>→INTEGER）。</para>
+///
+/// <para>🔴 <b>必须是带无参构造的可写类，不能写成 positional record</b>（2026-09-08 实测）：
+/// SqlSugar 用 <c>IDataReaderEntityBuilder.CreateBuilder</c> 反射构造结果对象，
+/// 对只有主构造函数的 record 会抛 <c>ArgumentNullException (Parameter 'con')</c>。
+/// 该异常被 <c>S0DimSameDbStagingLoader</c> 吞掉后静默回退逐行路径 ——
+/// 功能正确但性能优化全部失效，且日志里只有一行「判定异常」，极易被当成环境问题略过。</para>
+/// </summary>
+public sealed class S0DimSourceColumn
+{
+    /// <summary>SqlSugar 反射映射所需的无参构造。</summary>
+    public S0DimSourceColumn() { }
+
+    /// <summary>便于单测直接构造。</summary>
+    public S0DimSourceColumn(string name, string dataType, string columnType, bool isNullable)
+    {
+        Name = name;
+        DataType = dataType;
+        ColumnType = columnType;
+        IsNullable = isNullable;
+    }
+
+    /// <summary>物理列名（原样大小写）。</summary>
+    public string Name { get; set; } = "";
+
+    /// <summary><c>information_schema.COLUMNS.DATA_TYPE</c>，如 <c>varchar</c> / <c>tinyint</c>。</summary>
+    public string DataType { get; set; } = "";
+
+    /// <summary><c>information_schema.COLUMNS.COLUMN_TYPE</c>，如 <c>tinyint(1) unsigned zerofill</c>。</summary>
+    public string ColumnType { get; set; } = "";
+
+    /// <summary>是否可空。</summary>
+    public bool IsNullable { get; set; }
+}

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

@@ -183,12 +183,34 @@ public static class S0DimSqlBuilder
         $"SELECT {DimBizKeyExpr(def)} AS bk FROM {Q(def.DimTable)} {DimAlias} " +
         $"WHERE {DimAlias}.{Q(S0DimDefinition.TenantColumn)} = @TenantId";
 
-    /// <summary>集合差集计数：<paramref name="leftSql"/> \ <paramref name="rightSql"/>。</summary>
+    /// <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 COUNT(*) FROM ({leftSql}) x
-         LEFT JOIN ({rightSql}) y ON y.bk <=> x.bk
-         WHERE y.bk IS NULL
+         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>
@@ -312,4 +334,225 @@ public static class S0DimSqlBuilder
         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;
 }