-- 1.0.551.sql
--   P0-A 数据纠偏：精准清 std 种群 A + 保全 DWD 不可再生历史 + 清空重建 DWD + schema 收口。
--
--   ⚠️ 本脚本尚未在任何库执行（MIGRATION NOT EXECUTED）。
--      未执行原因是 Stop-Write Gate 未通过：S3 写入方仍在活跃跑批
--      （job_s3_mdp_sync_transform 小时级触发，job_smart_ops_kpi_mdp_bootstrap 随启动触发）。
--      执行前必须先取得真正的停写窗口，并确认 dwd_process_outsource_delivery
--      的 MAX(calc_time) 与 COUNT(*) 都不再前进。
--
--   号段说明（CLAUDE.md §二·7）：本文件曾以 WIP-P0A-CLEANUP.sql 占位；
--      rebase 到 origin/master 解决 1.0.536 撞号后先取 1.0.550，
--      随后与另一实例的 1.0.550（S8 菜单退役）实时撞号，本批让号顺延为 1.0.551。
--
-- ============================================================================
-- 背景（全部为只读实测结论，证据见本批审计回执）
--   dwd_process_outsource_delivery 曾混装两个业务对象：
--     种群 A = RoutingOpDetail（S0 标准工艺路线明细，主数据），无 work_order / po_no / po_line，
--              其 std 物化把 po_no/po_line 硬编码为 NULL；而 uk_work_op_stat 含这两列，
--              MySQL 唯一索引 NULL-distinct 语义使 ODKU 永不命中 → 每轮全量追加。
--     种群 B = PurOrdMaster(Potype='PW') JOIN PurOrdDetail，真实工序委外采购订单行。
--   代码侧止血已完成（S3_ROUTING_OUTSOURCE 入站注册与 std INSERT 已移除，DWD 投影已补
--   IFNULL(po_no,'') 使 po_no 不再为 NULL）。本脚本只处理存量。
--
--   Option A 目标粒度：一行 = 一条真实 PW 委外订单行在某 stat_date 的交付状态。
--   目标唯一键：(tenant_id, factory_id, stat_date, po_no, po_line)。
--
--   历史 stat_date 不可复现（@StatDate = 运行时墙钟日期，无回填入参），且 31 行对应的源单
--   已从 PurOrdMaster 物理删除 → 保全那 31 行就是历史恢复方案，不做"重新计算"。
-- ============================================================================
SET NAMES utf8mb4;

-- ===========================================================================
-- Step 0 · 前置门禁（fail-fast；任一不成立即中止，绝不带着错误前提往下做）
--   断言集合关系而非硬编码行数：行数会随跑批漂移，集合关系不会。
-- ===========================================================================

-- 种群 A 主谓词（与审计一致：批次来源 + biz_key 形态 + 结构特征三重冗余）
SET @std_total := (SELECT COUNT(*) FROM `mdp_std_process_outsource_order`);
SET @pop_a := (
  SELECT COUNT(*) FROM `mdp_std_process_outsource_order`
  WHERE `sync_batch_id`  LIKE 'S3\_MDP\_%'
    AND `source_biz_key` NOT LIKE '%|%'
    AND `work_order` = ''
    AND `po_no`   IS NULL
    AND `po_line` IS NULL
);
SET @survivors := (
  SELECT COUNT(*) FROM `mdp_std_process_outsource_order`
  WHERE NOT (`sync_batch_id` LIKE 'S3\_MDP\_%'
         AND `source_biz_key` NOT LIKE '%|%'
         AND `work_order` = ''
         AND `po_no`   IS NULL
         AND `po_line` IS NULL)
);
SELECT @std_total AS std_total, @pop_a AS population_a, @survivors AS survivors;

-- G1 · 集合完备：种群 A + 幸存者 必须恰好等于总量（不允许存在第三类）
SET @sql := IF(@pop_a + @survivors <> @std_total,
  "SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = 'P0A-G1: population_a + survivors <> std_total — predicate incomplete, abort'",
  'DO 0');
PREPARE s FROM @sql; EXECUTE s; DEALLOCATE PREPARE s;

-- G2 · 血缘交叉验证：主谓词命中集必须与 stg 血缘回溯集完全一致（双向差集为 0）
--      注意：不得在 EXISTS 里附加 source_system 等值条件 —— stg 侧两代分属
--      AIDOP(colon 形态) 与 AIDOPDEV_MYSQL(digits 形态)，而 std 侧被统一硬编码为 'AIDOP'，
--      加上该条件会假阴约 7.6 万行。
SET @main_not_prov := (
  SELECT COUNT(*) FROM `mdp_std_process_outsource_order` o
  WHERE o.`sync_batch_id` LIKE 'S3\_MDP\_%' AND o.`source_biz_key` NOT LIKE '%|%'
    AND o.`work_order` = '' AND o.`po_no` IS NULL AND o.`po_line` IS NULL
    AND NOT EXISTS (SELECT 1 FROM `mdp_stg_work_order_material` s
                    WHERE s.`source_table` = 'RoutingOpDetail'
                      AND s.`source_biz_key` = o.`source_biz_key`)
);
SET @prov_not_main := (
  SELECT COUNT(*) FROM `mdp_std_process_outsource_order` o
  WHERE EXISTS (SELECT 1 FROM `mdp_stg_work_order_material` s
                WHERE s.`source_table` = 'RoutingOpDetail'
                  AND s.`source_biz_key` = o.`source_biz_key`)
    AND NOT (o.`sync_batch_id` LIKE 'S3\_MDP\_%' AND o.`source_biz_key` NOT LIKE '%|%'
         AND o.`work_order` = '' AND o.`po_no` IS NULL AND o.`po_line` IS NULL)
);
SELECT @main_not_prov AS main_not_provenance, @prov_not_main AS provenance_not_main;
SET @sql := IF(@main_not_prov > 0 OR @prov_not_main > 0,
  "SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = 'P0A-G2: cleanup predicate disagrees with stg provenance — abort'",
  'DO 0');
PREPARE s FROM @sql; EXECUTE s; DEALLOCATE PREPARE s;

-- G3 · 幸存者必须全部是真实 PW 行（biz_key 形如 PurOrd|Line 且 po_no 非空）
SET @bad_survivor := (
  SELECT COUNT(*) FROM `mdp_std_process_outsource_order`
  WHERE NOT (`sync_batch_id` LIKE 'S3\_MDP\_%' AND `source_biz_key` NOT LIKE '%|%'
         AND `work_order` = '' AND `po_no` IS NULL AND `po_line` IS NULL)
    AND (`source_biz_key` NOT LIKE '%|%' OR IFNULL(`po_no`,'') = '')
);
SET @sql := IF(@bad_survivor > 0,
  "SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = 'P0A-G3: survivor set contains non-PW rows — abort'",
  'DO 0');
PREPARE s FROM @sql; EXECUTE s; DEALLOCATE PREPARE s;

-- G4 · overlap：种群 A 与真实 PW 集不得有交集
SET @overlap := (
  SELECT COUNT(*) FROM `mdp_std_process_outsource_order`
  WHERE `sync_batch_id` LIKE 'S3\_MDP\_%' AND `source_biz_key` NOT LIKE '%|%'
    AND `work_order` = '' AND `po_no` IS NULL AND `po_line` IS NULL
    AND `source_biz_key` LIKE '%|%'
);
SET @sql := IF(@overlap > 0,
  "SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = 'P0A-G4: population_a overlaps real PW rows — abort'",
  'DO 0');
PREPARE s FROM @sql; EXECUTE s; DEALLOCATE PREPARE s;

-- G5 · 待保全的 DWD 历史必须非空且 po_no/po_line 干净
--      ⚠️ 谓词必须是 IFNULL(po_no,'') <> '' 而不是 po_no IS NOT NULL：
--      代码侧 IFNULL 补丁上线后，种群 A 的行 po_no 变成**空串**而非 NULL，
--      用 IS NOT NULL 会把这些垃圾一并当成"不可再生历史"保下来。
SET @preserve_cnt := (SELECT COUNT(*) FROM `dwd_process_outsource_delivery` WHERE IFNULL(`po_no`,'') <> '');
SELECT @preserve_cnt AS dwd_rows_to_preserve;
SET @sql := IF(@preserve_cnt = 0,
  "SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = 'P0A-G5: no real-PO rows found in DWD — refuse to truncate blindly'",
  'DO 0');
PREPARE s FROM @sql; EXECUTE s; DEALLOCATE PREPARE s;

-- ===========================================================================
-- Step 1 · 保全 DWD 不可再生历史 + 元数据留证
--   只备份真实 PO 子集（约 31 行），不复制 3269 万行错误派生数据：
--   零消费者 / 零 FK / 可重建（今后）/ 历史上已 TRUNCATE 过 / 权威源仍在。
-- ===========================================================================
CREATE TABLE IF NOT EXISTS `dwd_process_outsource_delivery_keep_p0a` LIKE `dwd_process_outsource_delivery`;
INSERT IGNORE INTO `dwd_process_outsource_delivery_keep_p0a`
SELECT * FROM `dwd_process_outsource_delivery` WHERE IFNULL(`po_no`,'') <> '';

-- 保全完整性断言：备份行数必须等于源侧真实 PO 行数
SET @kept := (SELECT COUNT(*) FROM `dwd_process_outsource_delivery_keep_p0a`);
SET @sql := IF(@kept <> @preserve_cnt,
  "SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = 'P0A-S1: preserved row count mismatch — abort before truncate'",
  'DO 0');
PREPARE s FROM @sql; EXECUTE s; DEALLOCATE PREPARE s;

-- 元数据留证（复用 1.0.350 既有范式与表）
CREATE TABLE IF NOT EXISTS `aidop_migration_derived_cleanup_log` (
  `id` BIGINT NOT NULL AUTO_INCREMENT,
  `migration_version` VARCHAR(32) NOT NULL,
  `table_name` VARCHAR(128) NOT NULL,
  `estimated_rows` BIGINT NOT NULL DEFAULT 0,
  `data_length` BIGINT NOT NULL DEFAULT 0,
  `reason` VARCHAR(500) NOT NULL,
  `captured_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
  PRIMARY KEY (`id`),
  UNIQUE KEY `uk_migration_cleanup` (`migration_version`,`table_name`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

INSERT INTO `aidop_migration_derived_cleanup_log`
(`migration_version`,`table_name`,`estimated_rows`,`data_length`,`reason`)
SELECT '1.0.551','dwd_process_outsource_delivery',IFNULL(TABLE_ROWS,0),IFNULL(DATA_LENGTH,0),
       CONCAT('P0-A：RoutingOpDetail 主数据被误建模为委外交付事实，叠加 uk 含可空 po_no 致 ODKU 永不命中，膨胀至 3269 万行；',
              '真实 PO 子集 ', @preserve_cnt, ' 行已存入 dwd_process_outsource_delivery_keep_p0a 并于 schema 收口后回插；',
              '其余为错误派生数据，零消费者、零 FK、可由 std 重新生成，不做全量备份')
FROM information_schema.TABLES
WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='dwd_process_outsource_delivery'
ON DUPLICATE KEY UPDATE `reason`=VALUES(`reason`), `estimated_rows`=VALUES(`estimated_rows`), `data_length`=VALUES(`data_length`);

-- ===========================================================================
-- Step 2 · 精准清 std 种群 A（保留真实 PW；不得 TRUNCATE 整表）
--   备份沿用 1.0.351/352/353 的范式 A：CREATE TABLE LIKE + INSERT IGNORE + DELETE。
-- ===========================================================================
CREATE TABLE IF NOT EXISTS `mdp_std_process_outsource_order_bak_p0a` LIKE `mdp_std_process_outsource_order`;
INSERT IGNORE INTO `mdp_std_process_outsource_order_bak_p0a`
SELECT * FROM `mdp_std_process_outsource_order`
WHERE `sync_batch_id`  LIKE 'S3\_MDP\_%'
  AND `source_biz_key` NOT LIKE '%|%'
  AND `work_order` = ''
  AND `po_no`   IS NULL
  AND `po_line` IS NULL;

DELETE FROM `mdp_std_process_outsource_order`
WHERE `sync_batch_id`  LIKE 'S3\_MDP\_%'
  AND `source_biz_key` NOT LIKE '%|%'
  AND `work_order` = ''
  AND `po_no`   IS NULL
  AND `po_line` IS NULL;

-- 清理后立即断言：种群 A 归零，且幸存者数量与门禁阶段一致
SET @pop_a_after := (
  SELECT COUNT(*) FROM `mdp_std_process_outsource_order`
  WHERE `sync_batch_id` LIKE 'S3\_MDP\_%' AND `source_biz_key` NOT LIKE '%|%'
    AND `work_order` = '' AND `po_no` IS NULL AND `po_line` IS NULL
);
SET @std_after := (SELECT COUNT(*) FROM `mdp_std_process_outsource_order`);
SELECT @pop_a_after AS population_a_after, @std_after AS std_total_after;
SET @sql := IF(@pop_a_after <> 0 OR @std_after <> @survivors,
  "SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = 'P0A-S2: std cleanup result mismatch — abort before DWD truncate'",
  'DO 0');
PREPARE s FROM @sql; EXECUTE s; DEALLOCATE PREPARE s;
-- 注：tenant 0 / 1300000000001 的真实 PW 行仍保留在 std。
--     "进不了 DWD"（被 CHECK 拦）不等于"应该从 std 删除"。

-- ===========================================================================
-- Step 3 · 清空 DWD
--   选 TRUNCATE 而非分批 DELETE：错误数据无业务价值、零 FK、零消费者，
--   且 DELETE 会产生约 5-8GB binlog（ROW+FULL image）与数十 GB undo，
--   却**不把空间归还 OS**；TRUNCATE 立即归还约 7.9GB。
-- ===========================================================================
TRUNCATE TABLE `dwd_process_outsource_delivery`;

-- ===========================================================================
-- Step 4 · 空表期 schema 收口（此时行数为 0，DDL 代价最低且不会因存量 NULL 失败）
-- ===========================================================================

-- 4.1 po_no / po_line → NOT NULL DEFAULT ''（幂等：仅当当前仍 nullable 时执行）
SET @po_no_nullable := (SELECT IS_NULLABLE FROM information_schema.COLUMNS
  WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='dwd_process_outsource_delivery' AND COLUMN_NAME='po_no');
SET @sql := IF(@po_no_nullable = 'YES',
  "ALTER TABLE `dwd_process_outsource_delivery` MODIFY COLUMN `po_no` VARCHAR(100) NOT NULL DEFAULT ''",
  'DO 0');
PREPARE s FROM @sql; EXECUTE s; DEALLOCATE PREPARE s;

SET @po_line_nullable := (SELECT IS_NULLABLE FROM information_schema.COLUMNS
  WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='dwd_process_outsource_delivery' AND COLUMN_NAME='po_line');
SET @sql := IF(@po_line_nullable = 'YES',
  "ALTER TABLE `dwd_process_outsource_delivery` MODIFY COLUMN `po_line` VARCHAR(50) NOT NULL DEFAULT ''",
  'DO 0');
PREPARE s FROM @sql; EXECUTE s; DEALLOCATE PREPARE s;

-- 4.2 唯一键改为 Option A 粒度：(tenant_id, factory_id, stat_date, po_no, po_line)
--     旧键 uk_work_op_stat 含 work_order/op_code，对 RoutingOpDetail 来源无区分度。
SET @has_old_uk := (SELECT COUNT(*) FROM information_schema.STATISTICS
  WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='dwd_process_outsource_delivery' AND INDEX_NAME='uk_work_op_stat');
SET @sql := IF(@has_old_uk > 0,
  'ALTER TABLE `dwd_process_outsource_delivery` DROP INDEX `uk_work_op_stat`',
  'DO 0');
PREPARE s FROM @sql; EXECUTE s; DEALLOCATE PREPARE s;

SET @has_new_uk := (SELECT COUNT(*) FROM information_schema.STATISTICS
  WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='dwd_process_outsource_delivery' AND INDEX_NAME='uk_po_line_stat');
SET @sql := IF(@has_new_uk = 0,
  'ALTER TABLE `dwd_process_outsource_delivery` ADD UNIQUE INDEX `uk_po_line_stat` (`tenant_id`,`factory_id`,`stat_date`,`po_no`,`po_line`)',
  'DO 0');
PREPARE s FROM @sql; EXECUTE s; DEALLOCATE PREPARE s;

-- 4.3 租户守卫 CHECK：按**实际名**处理，不得沿用 1.0.353 声明的名字
--     实测漂移：1.0.353 声明 ck_dwd_process_outsource_valid_tenant，
--     但该表在 1.0.353 之后被 CodeFirst 重建过，约束被匿名重建为
--     dwd_process_outsource_delivery_chk_1（全库唯一的自动命名 CHECK）。
--     此处按语义（CHECK_CLAUSE 含 tenant_id 且含 1300000000001）定位，统一收敛到稳定名。
SET @tenant_chk_cnt := (
  SELECT COUNT(*) FROM information_schema.CHECK_CONSTRAINTS cc
  JOIN information_schema.TABLE_CONSTRAINTS tc
    ON tc.CONSTRAINT_SCHEMA=cc.CONSTRAINT_SCHEMA AND tc.CONSTRAINT_NAME=cc.CONSTRAINT_NAME
  WHERE tc.TABLE_SCHEMA=DATABASE() AND tc.TABLE_NAME='dwd_process_outsource_delivery'
    AND tc.CONSTRAINT_TYPE='CHECK'
    AND cc.CHECK_CLAUSE LIKE '%tenant_id%' AND cc.CHECK_CLAUSE LIKE '%1300000000001%'
);
SET @sql := IF(@tenant_chk_cnt > 1,
  "SIGNAL SQLSTATE '45000' SET MESSAGE_TEXT = 'P0A-S4: duplicate tenant-guard CHECK constraints found — manual review required'",
  'DO 0');
PREPARE s FROM @sql; EXECUTE s; DEALLOCATE PREPARE s;

SET @tenant_chk_name := (
  SELECT tc.CONSTRAINT_NAME FROM information_schema.CHECK_CONSTRAINTS cc
  JOIN information_schema.TABLE_CONSTRAINTS tc
    ON tc.CONSTRAINT_SCHEMA=cc.CONSTRAINT_SCHEMA AND tc.CONSTRAINT_NAME=cc.CONSTRAINT_NAME
  WHERE tc.TABLE_SCHEMA=DATABASE() AND tc.TABLE_NAME='dwd_process_outsource_delivery'
    AND tc.CONSTRAINT_TYPE='CHECK'
    AND cc.CHECK_CLAUSE LIKE '%tenant_id%' AND cc.CHECK_CLAUSE LIKE '%1300000000001%'
  LIMIT 1
);
-- 若现存名不是稳定名，则丢弃后按稳定名重建（空表期，代价可忽略）
SET @sql := IF(@tenant_chk_name IS NOT NULL AND @tenant_chk_name <> 'ck_dwd_process_outsource_valid_tenant',
  CONCAT('ALTER TABLE `dwd_process_outsource_delivery` DROP CHECK `', @tenant_chk_name, '`'),
  'DO 0');
PREPARE s FROM @sql; EXECUTE s; DEALLOCATE PREPARE s;

SET @has_stable_chk := (SELECT COUNT(*) FROM information_schema.TABLE_CONSTRAINTS
  WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME='dwd_process_outsource_delivery'
    AND CONSTRAINT_TYPE='CHECK' AND CONSTRAINT_NAME='ck_dwd_process_outsource_valid_tenant');
SET @sql := IF(@has_stable_chk = 0,
  'ALTER TABLE `dwd_process_outsource_delivery` ADD CONSTRAINT `ck_dwd_process_outsource_valid_tenant` CHECK (`tenant_id` NOT IN (0,1,1300000000001))',
  'DO 0');
PREPARE s FROM @sql; EXECUTE s; DEALLOCATE PREPARE s;

-- ===========================================================================
-- Step 5 · 回插保全的历史
--   显式排除非法租户（不依赖隐式 filter），并显式要求 po_no/po_line 非空。
-- ===========================================================================
INSERT INTO `dwd_process_outsource_delivery`
SELECT * FROM `dwd_process_outsource_delivery_keep_p0a`
WHERE `tenant_id` NOT IN (0,1,1300000000001)
  AND IFNULL(`po_no`,'')   <> ''
  AND IFNULL(`po_line`,'') <> '';

SELECT COUNT(*) AS dwd_rows_after_restore FROM `dwd_process_outsource_delivery`;
SELECT `estimated_rows` AS rows_before_cleanup FROM `aidop_migration_derived_cleanup_log`
WHERE `migration_version`='1.0.551' AND `table_name`='dwd_process_outsource_delivery';

-- ===========================================================================
-- 回滚（注释，非可执行；执行前请确认无消费者）
--   std：INSERT IGNORE INTO mdp_std_process_outsource_order
--          SELECT * FROM mdp_std_process_outsource_order_bak_p0a;
--   dwd：错误派生数据不可回滚也不需要回滚（零消费者、可由 std 重新生成）；
--        真实历史由 dwd_process_outsource_delivery_keep_p0a 保全，可再次回插。
--   schema：ALTER ... MODIFY po_no/po_line 恢复 nullable；
--           DROP INDEX uk_po_line_stat; ADD UNIQUE uk_work_op_stat(...原列序...)。
--   ⚠️ 回滚 schema 会让 ODKU 重新失效、膨胀复发，仅在确认整改方案作废时执行。
-- ===========================================================================