check_subset.py
13.3 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
重名表「丢弃份是不是保留份的子集」精确校验 —— 六库合一的最后一道数据闸门。
## 为什么需要它
[`merge-conflict-plan.md`](../merge-conflict-plan.md) §0 列的**必须逐一确认的例外**有两条，
其中第 2 条是：
> 两侧行数都非 0 且差异大时，不能假定"主份已包含副本全部行"——
> 要抽样比对，确认没有只存在于副本的历史数据。
抽样比对只能给出"抽样命中"这种**概率性**结论（§2.1 里 `park_in_out_record`
那次是抽 300 个回查、命中 300/300）。本工具把这件事做成**精确**的：
所有六个库都在**同一台实例**上（见 [`legacy-db-actual.md`](../legacy-db-actual.md) §1），
所以可以直接发跨库 `LEFT JOIN`，一次问清"丢弃份里有多少行的主键在保留份里**找不到**"。
那个数就是**合库后会真正丢掉的行数**。0 ⇒ 这份可以安全丢弃；非 0 ⇒ 必须人工处置
（取并集 / 查清调用方 / 单独导出）。
## ⚠️ 只读且只看键列
本脚本**只发 SELECT**，且只 SELECT **主键列**（外加显式配置的键列）——
绝不 `SELECT *`。这一条是刻意的：清单里有些表（如 `td_b_wx_config`）
装着微信支付配置，通配查询会把密钥打进终端和日志。只投影键列，
就**不可能**有任何敏感列值出现在输出里。
## ⚠️ 键选错 = 假警报（本脚本踩过的坑）
`--key` 用来把判定从「主键」换成「业务键」，**副本换过 id 空间时，只有业务键才说真话**。
但反过来也成立：**业务键选错会造出成片假独有**。已修掉的两例：
- `t_vip_card_cust_info`：按主键 `id` 比 → 27 行"独有"；按 `card_no` 比 → **0**。
同一批卡换了 id，不是丢了数据
- `td_b_wx_config`：按 `org_id` 比 → 18 行"独有"；按 `appid` 比 → **3**。
老库的 `org_id` **大量为 NULL**，而 `NULL = NULL` 在 SQL 里不成立，
于是 join 必然落空 —— 那 18 行是**连接失败**，不是数据缺失。
这张表的真实查询键是 `appid`：`WxPayApiConfigKit` 就是按 appId 找配置的
（找不到直接抛 `IllegalStateException`）
> 规律：**键列里只要可能有 NULL，用它做 join 就一定会误报**。
> 选业务键前先问一句「这个列在老数据里是否普遍为空」。
用法：
python3 -I sql/tools/check_subset.py --env 私密/legacy-ro.env
python3 -I sql/tools/check_subset.py --env 私密/legacy-ro.env --only td_c_cust_person
python3 -I sql/tools/check_subset.py --env 私密/legacy-ro.env --sample 20
退出码：0 = 所有待丢弃份都已被证明是子集（或已被人工判定）；1 = 仍有需人工处置的。
"""
import argparse
import os
import sys
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from legacy_db import connect, db_names, load_env # noqa: E402
# ---------------------------------------------------------------------------
# 待校验清单：唯一的事实来源
#
# 每项 = (表, 保留侧模块, 丢弃侧模块, 显式键列 or None 表示自动取主键, 备注)
# 模块名对应 env 文件里的 LEGACY_DB_<模块>；注意历史填写里
# PAY → zt_order（pay 域活库 + 2018 订单归档）
# ORDER → zt_order_n（order 域活库）
# —— 库名与业务域是错位的，所以这里必须写「模块」而不是直接写库名，
# 由 db_names() 解析，避免把错位的名字抄错第二次。
#
# ⚠️ 第 4 列**不要留 None 就完事**：副本换过 id 空间时主键会满屏假独有。
# 已固化三处业务键（见文件头「键选错 = 假警报」）：
# t_vip_card_cust_info → card_no sys_org → org_code
# td_b_wx_config → appid （org_id 为 NULL 会造出 18 行假独有）
# ---------------------------------------------------------------------------
CHECKS = [
# ---- §2.1 「主份是超集且活跃」5 组：本工具要给每份副本一个精确的"可丢"证明 ----
("parking_lot", "PARK", "BILL", "pl_no", "zt_bill 1603 行（无更新）"),
("parking_lot", "PARK", "PAY", "pl_no", "zt_order 1465 行（无更新）"),
("parking_lot", "PARK", "ORDER", "pl_no", "zt_order_n 1979 行（无更新）"),
("park_in_out_record", "PARK", "PAY", "order_id",
"§2.1 已抽 300/300 命中，此处改做精确全量"),
("t_vip_card_cust_info", "BILL", "PAY", "card_no",
"zt_order 1892 行陈旧；按 id 会假报 27 行，按 card_no 为 0"),
("tp_p_berths", "PARK", "ORDER", None,
"zt_order_n 20554 行；按 id 有 31 行独有，但 1681 行同 id 已被改写"),
("tp_e_employee_charge", "PARK", "ORDER", None, "zt_order_n 940 行"),
# ---- §2.2 🔴「动手前必须查调用方」4 组 ----
("sys_org", "SYS", "PARK", "org_code", "zt_park 20 行（残缺副本）"),
("sys_org", "SYS", "ORDER", "org_code", "zt_order_n 20 行（残缺副本）"),
("sys_org", "SYS", "PAY", "org_code", "zt_order 0 行"),
("td_c_cust_person", "PAY", "ORDER", "cust_id", "方向相反：活的在 zt_order（超集）"),
("td_c_coupon_person", "PAY", "ORDER", None, "zt_order_n 5720 行待定性"),
("td_b_settlement_record", "ORDER", "PAY", None, "§2.2：zt_order 90 行建议并集"),
# ---- §1 「结构完全一致」7 组：结构同，但数据来源仍要定 ----
("income_temp", "ORDER", "PARK", None, "353/353 行，两侧同数"),
("month_dchz", "ORDER", "PARK", "time,pl_no", "199/199 行，两侧同数；两侧都无主键"),
("table_15_free", "ORDER", "PARK", "dateStr,pl_no", "344/342 行，差 2；两侧都无主键"),
("td_b_settlement", "ORDER", "PAY", None, "两侧皆空"),
("td_b_settlement_org", "ORDER", "PAY", None, "zt_order 0 行"),
("td_b_settlement_serial", "ORDER", "PAY", None, "zt_order 0 行"),
("td_b_settlement_share_record", "ORDER", "PAY", None, "20/33 行，两侧都有数据"),
("td_p_parking_share_order_temp", "ORDER", "PAY", None, "临时表"),
# ---- §2.3 🟡 无活跃方 3 组 ----
("tb_b_park_income_day", "ORDER", "PAY", None, "zt_order 1796 行待定性"),
("td_b_wx_config", "ORDER", "PAY", "appid",
"两侧 27/22 行且都无更新时间；⚠️ 本表含支付配置，仅比对 appid，不比对内容"),
]
def pk_columns(cur, db, table):
"""主键列（按索引顺序）。没有主键返回 []。"""
cur.execute(
"SELECT COLUMN_NAME FROM information_schema.STATISTICS "
"WHERE TABLE_SCHEMA=%s AND TABLE_NAME=%s AND INDEX_NAME='PRIMARY' "
"ORDER BY SEQ_IN_INDEX", (db, table))
return [r[0] for r in cur.fetchall()]
def one(cur, db, table):
cur.execute("SELECT COUNT(*) FROM `%s`.`%s`" % (db, table))
return cur.fetchone()[0]
def check(cur, table, keep_db, drop_db, key, sample):
"""返回 dict 结论。只投影键列，不做 SELECT *。"""
# 键列：显式给的优先，否则自动取「丢弃侧」的主键（要丢的那份才是被探测的对象）
if key:
cols = [c.strip() for c in key.split(",")]
else:
cols = pk_columns(cur, drop_db, table)
res = {"table": table, "keep_db": keep_db, "drop_db": drop_db,
"key": cols, "keep_n": None, "drop_n": None, "drop_only": None,
"samples": [], "note": ""}
if not cols:
res["note"] = "两侧都没有主键，且未配置键列 → 无法自动判定"
return res
keep_pk = pk_columns(cur, keep_db, table)
missing_in_keep = [c for c in cols if c not in keep_pk]
if missing_in_keep:
# 键不是保留侧主键时的两点说明：
# ① 结论强度下降：只能说「这个键值在保留侧不存在」，不能说「这一行不存在」；
# ② 若该键在**丢弃侧**也不唯一，行数会把「同一个业务对象」重复计数，
# 所以下面同时给出行数与**去重后的键数**，两者不等就说明键不唯一。
res["note"] = ("键列 %s 在保留侧不是主键 → 结论是「键值不存在」而非「行不存在」"
% ",".join(missing_in_keep))
on = " AND ".join("k.`%s` = d.`%s`" % (c, c) for c in cols)
res["keep_n"] = one(cur, keep_db, table)
res["drop_n"] = one(cur, drop_db, table)
distinct = ("COUNT(DISTINCT d.`%s`)" % cols[0] if len(cols) == 1 else
"COUNT(DISTINCT CONCAT_WS('\\x1f', %s))"
% ", ".join("d.`%s`" % c for c in cols))
cur.execute(
"SELECT COUNT(*), %s FROM `%s`.`%s` d LEFT JOIN `%s`.`%s` k ON %s "
"WHERE k.`%s` IS NULL"
% (distinct, drop_db, table, keep_db, table, on, cols[0]))
res["drop_only"], res["drop_only_distinct"] = cur.fetchone()
if res["drop_only"]:
cur.execute(
"SELECT %s FROM `%s`.`%s` d LEFT JOIN `%s`.`%s` k ON %s "
"WHERE k.`%s` IS NULL LIMIT %d"
% (", ".join("d.`%s`" % c for c in cols),
drop_db, table, keep_db, table, on, cols[0], sample))
res["samples"] = cur.fetchall()
return res
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--env", default=os.path.join(
os.path.dirname(os.path.abspath(__file__)), "..", "..", "私密", "legacy-ro.env"))
ap.add_argument("--only", action="append",
help="只跑这几张表（可重复；默认全部）")
ap.add_argument("--sample", type=int, default=10,
help="发现独有行时最多列出几条键（默认 10）")
ap.add_argument("--timeout", type=int, default=600,
help="单条 SQL 的读超时秒数（默认 600；大表逐行探测需要）")
ap.add_argument("--key", action="append", metavar="表=列1,列2",
help="覆盖某张表的键（按表名匹配，可重复）。"
"用途：按主键判定「行不存在」后，再按**业务键**判定一次 —— "
"副本若只是换了 id 空间，用业务键才不会误判成「丢了数据」")
a = ap.parse_args()
cfg = load_env(a.env)
dbs = db_names(cfg)
# --key 的覆盖：只按表名匹配，允许同一张表在不同保留/丢弃组合上都换键
over = {}
for kv in (a.key or []):
t, _, cols = kv.partition("=")
if not cols:
print("--key 格式应为 表=列[,列]", file=sys.stderr)
return 2
over[t.strip()] = cols.strip()
want = [c for c in CHECKS if not a.only or c[0] in a.only]
if over:
# 显式给了键的表，即使没在 CHECKS 的 --only 里也要跑，避免"改了键却没跑"
seen = {c[0] for c in want}
for t in over:
if t not in seen:
want += [c for c in CHECKS if c[0] == t]
if not want:
print("没有匹配的检查项", file=sys.stderr)
return 2
conn = connect(cfg, read_timeout=a.timeout)
need_manual = []
try:
with conn.cursor() as cur:
print("=" * 78)
print("重名表「丢弃份 ⊂ 保留份」精确校验（跨库 LEFT JOIN，只看键列）")
print("=" * 78)
for table, keep_mod, drop_mod, key, note in want:
kd, dd = dbs[keep_mod], dbs[drop_mod]
r = check(cur, table, kd, dd, over.get(table, key), a.sample)
head = "%s ← 保留 %s(%s) / 丢弃 %s(%s)" % (table, keep_mod, kd, drop_mod, dd)
print("\n" + head)
print(" 备注: %s" % note)
if r["keep_n"] is None:
print(" ✋ %s" % r["note"])
need_manual.append(head)
continue
print(" 保留侧 %d 行 / 丢弃侧 %d 行 / 键 %s"
% (r["keep_n"], r["drop_n"], "+".join(r["key"])))
if r["note"]:
print(" ⚠️ %s" % r["note"])
if r["drop_only"] == 0:
print(" ✅ 丢弃侧**全部**落在保留侧内 → 可安全丢弃（丢 0 行）")
else:
uniq = ""
if r["drop_only_distinct"] != r["drop_only"]:
uniq = ("（去重后 %d 个键 —— 键在丢弃侧不唯一，"
"上面是行数）" % r["drop_only_distinct"])
print(" 🔴 丢弃侧有 **%d 行**在保留侧找不到%s → 丢掉会真的丢数据"
% (r["drop_only"], uniq))
if r["samples"]:
print(" 独有键样本: %s"
% ", ".join(str(s[0]) if len(s) == 1 else str(s)
for s in r["samples"]))
need_manual.append(head)
finally:
conn.close()
print("\n" + "=" * 78)
if need_manual:
print("需人工处置 %d 组：" % len(need_manual))
for h in need_manual:
print(" · %s" % h)
print("\n处置方向：独有行取并集（`INSERT ... SELECT` 补进保留份，再丢弃副本）；"
"\n 无主键无法判定的，先给它定一个业务键再跑一次。")
else:
print("✅ 全部待丢弃份都已被证明是保留份的子集 —— 可安全合并。")
print("=" * 78)
return 1 if need_manual else 0
if __name__ == "__main__":
sys.exit(main())