-- =====================================================================================
-- 1.0.521 · S1 · 让 mdp_std_so 成为销售订单行的真实镜像（淘汰源侧已删除的孤儿）
--
-- ── 缺陷 ────────────────────────────────────────────────────────────────────────────
-- 贴源层 mdp_stg_so 是纯 upsert、从不淘汰行；标准层 mdp_std_so 的 INSERT 又不带任何
-- 批次条件、每轮重读整张贴源表。于是源侧被硬删除的订单行会：
--   ① 永远留在 mdp_stg_so；
--   ② 每轮被重新物化进 mdp_std_so，并且 sync_batch_id / sync_time 被刷成最新
--      —— 所以「看时间戳」根本认不出它们是陈旧数据。
--
-- 实测（aidopdev，2026-09-08）租户 797403760988229：
--   mdp_std_so 414 行 = 28 行真实存活 + 25 行 biz_key 旧分隔符(':')的重复行
--                      + 361 行源侧已删除的孤儿
--   其余三个租户 339/14/9 行，与源侧逐行一致，孤儿为 0。
--
-- ── 影响面不止 S8 ───────────────────────────────────────────────────────────────────
-- mdp_std_so 有 16 个读取点，且没有任何一个带「当前行」判据：
--   · dwd_ship_trans 以它为驱动表 —— 每条孤儿凭空造出一条发运行，日期早已过期，
--     必然被判 DELAYED / 高风险；
--   · S1_L1_001/002/003、S1_L2_010/011/012、S7_L1_001/002/003、S9_L1_003 等 KPI
--     的分子分母都被孤儿污染；
--   · S8 Rule02（订单交付延期预警）若改读中台，孤儿会直接变成凭空的异常单。
--
-- ── 修法 ────────────────────────────────────────────────────────────────────────────
-- 不新增「当前标记」列，而是让标准层恢复「镜像」语义，这样 16 个消费者一个都不用改：
--   ① 本脚本把 S1_SEORDER / S1_SEORDER_ENTRY 的 sync_mode 由 INCR 改为 FULL；
--   ② 代码侧（S1MdpSyncTransformService 的 mdp_std_so 构建）只取每个租户的
--      最新一次 sync_batch_id；
--   ③ 本脚本一次性清掉标准层里已经沉淀的孤儿。
--
-- 为什么 ① 是 ② 成立的前提：MdpSyncWindowResolver.Apply 对 sync_mode='FULL' 的实体
-- 会强制 ctx.FullRefresh=true（:34-35），因而在**任何**调用路径上都跳过增量水位——
-- 包括 RunInboundAsync 与 MdpHotWatchService 这两条 fullRefresh=false 的路径。
-- 若维持 INCR，那两条路径会产出「只含变更行」的部分批次，② 会把未变更的存量行
-- 当成孤儿删掉，后果比现在的缺陷更严重。
--
-- 代价可忽略：这两个实体实测分别只有 ~390 / ~81 行，且 S1 全量作业本来就每轮
-- 以 fullRefresh:true 全量重读它们 —— 本次只是让「声明」与「实际行为」一致。
-- =====================================================================================

-- ── ① 让声明与实际行为一致：这两个实体每轮都是全量重读 ─────────────────────────────
UPDATE `mdp_entity`
   SET `sync_mode` = 'FULL',
       `incr_column` = NULL,
       `update_time` = NOW()
 WHERE `entity_code` IN ('S1_SEORDER', 'S1_SEORDER_ENTRY')
   AND `sync_mode` <> 'FULL';

-- ── ② 一次性清除标准层已沉淀的孤儿 ──────────────────────────────────────────────────
-- 判据：该 (tenant_id, source_biz_key) 不在贴源层「该租户最新一次 sync_batch_id」里。
--
-- ⚠️ 安全闸门 —— 「该租户在贴源层确实有数据」：
--    若某租户在 mdp_stg_so 里一行都没有（从未同步 / 贴源被清过），
--    它在下面的 lb2 里不会产生分组行，JOIN 直接把该租户的标准层整体排除在删除范围外。
--    这与旧写法的 `EXISTS (...)` 闸门等价，但由 JOIN 天然承担，不必单列一个子查询。
--    这里宁可不删，也不冒清空的风险。
--
-- ── 为什么用派生表而不是逐行相关子查询（2026-09-08 性能修复）─────────────────────────
-- 原写法把「取该租户最新批次」写成**每一行都重算一次**的相关子查询：
--   NOT EXISTS(... lb.sync_batch_id = (SELECT ... ORDER BY sync_time DESC, id DESC LIMIT 1))
-- EXPLAIN 实测（共享库）：g 全表扫 38722 行、lb 全表扫 38722 行，
-- 最内层 x 是 DEPENDENT SUBQUERY + Using filesort —— 即「每条 std 行都对贴源做一次排序」。
-- 只读实测该判据取数 51.8s / 386 行；一旦贴源继续增长就会撞上
-- AutoVersionUpdate.MigrationCommandTimeoutSeconds = 600s，重演 1.0.517 的启动死循环。
--
-- 改法只动查询形状，不动业务语义：把贴源侧「订单行」这一段提成 CTE `stg` 物化一次，
-- 「每租户最新批次」与「最新批次里有哪些 biz_key」都从这一份物化结果里取。
-- EXPLAIN 实测（共享库 8.0.31 与本地 8.0.44 计划一致）：mdp_stg_so 只被扫一次，
-- CTE 被两处以 ref + auto_key 复用，DEPENDENT SUBQUERY 与逐行 filesort 全部消失。
--
-- 取「最新批次」用 ROW_NUMBER() 而不是 GROUP_CONCAT + SUBSTRING_INDEX：
-- 后者要求整个拼接串不超过 group_concat_max_len（实测该值 1024，
-- 而单租户最多 414 行 × 45 字符 ≈ 18630 字符，必然触发尾部截断并产生 warning 1260）。
-- 虽然我们只取第一个元素、截断发生在尾部因而结果仍正确，
-- 但让一条删除语句的正确性依赖「截断位置」太脆弱。ROW_NUMBER 是精确的，没有这个前提。
-- 等价性已实测：新旧两种写法在共享库与本地库上逐租户取到的 sync_batch_id 完全一致
-- （mismatch = 0），且待删行集合逐行相同（386 / 386，only_old = 0，only_new = 0）。
WITH stg AS (
  SELECT `tenant_id`, `source_biz_key`, `sync_batch_id`,
         ROW_NUMBER() OVER (PARTITION BY `tenant_id`
                            ORDER BY `sync_time` DESC, `id` DESC) AS `rn`
    FROM `mdp_stg_so`
   WHERE `source_table` = 'crm_seorderentry'
)
DELETE s FROM `mdp_std_so` s
  JOIN (SELECT `tenant_id`, `sync_batch_id` AS `latest_batch`
          FROM stg WHERE `rn` = 1) lb2
    ON lb2.`tenant_id` = s.`tenant_id`
 WHERE s.`source_table` = 'crm_seorderentry'
   AND NOT EXISTS (
        SELECT 1 FROM stg lb
         WHERE lb.`tenant_id` = s.`tenant_id`
           AND lb.`source_biz_key` = s.`source_biz_key`
           AND lb.`sync_batch_id` = lb2.`latest_batch`);