123 lines
4.5 KiB
Python
123 lines
4.5 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""DDL 幂等修正器：把 tenant_id 提升为每张表的物理首列（QC 第三轮 #1 配套）。
|
||
|
||
背景
|
||
----
|
||
data-model.md 的「tenant_id 强制打头」有两层含义：
|
||
(a) 键前缀契约 —— 每个复合 UNIQUE KEY / KEY 的首列必须是 tenant_id（影响分区裁剪与查询计划，**必须成立**）；
|
||
(b) 物理列序 —— CREATE TABLE 的第一个列定义是 tenant_id（可读性/评审口径）。
|
||
核验器 gate_64 以 (a) 为判定项、(b) 为 INFO。本脚本把 (b) 也补齐，使两种口径同时成立，
|
||
且**幂等**：已经是首列的表不做任何改动，重复执行结果一致。
|
||
|
||
用法
|
||
----
|
||
python3 apps/pbls/scripts/patch_ddl_tenant_first.py \
|
||
--file apps/pbls/scripts/ddl/pbls_tables.sql [--dry-run]
|
||
|
||
安全约束
|
||
--------
|
||
* 只做「列定义整行搬移」，不改列类型、不改索引、不改表选项、不改注释；
|
||
* 解析失败（找不到 tenant_id 列 / 括号不配对）时跳过该表并打印 SKIP，绝不写坏文件；
|
||
* --dry-run 只报告将变更的表与行号，不落盘。
|
||
"""
|
||
|
||
import argparse
|
||
import io
|
||
import os
|
||
import re
|
||
import sys
|
||
|
||
CREATE_RX = re.compile(r"CREATE\s+TABLE\s+(?:IF\s+NOT\s+EXISTS\s+)?[`\"]?(\w+)[`\"]?\s*\(", re.I)
|
||
TENANT_COL_RX = re.compile(r"^\s*[`\"]?tenant_id[`\"]?\s+", re.I)
|
||
|
||
|
||
def match_paren(text, open_idx):
|
||
depth = 0
|
||
for i in range(open_idx, len(text)):
|
||
if text[i] == "(":
|
||
depth += 1
|
||
elif text[i] == ")":
|
||
depth -= 1
|
||
if depth == 0:
|
||
return i
|
||
return -1
|
||
|
||
|
||
def line_of(text, offset):
|
||
return text[:offset].count("\n") + 1
|
||
|
||
|
||
def patch(sql):
|
||
"""返回 (new_sql, changes, skips)。changes/skips 元素为 (表名, 行号, 说明)。"""
|
||
changes, skips = [], []
|
||
out = sql
|
||
offset_shift = 0
|
||
for m in CREATE_RX.finditer(sql):
|
||
tbl = m.group(1)
|
||
open_idx = m.end() - 1
|
||
close_idx = match_paren(sql, open_idx)
|
||
if close_idx < 0:
|
||
skips.append((tbl, line_of(sql, m.start()), "括号不配对，跳过"))
|
||
continue
|
||
body = sql[open_idx + 1:close_idx]
|
||
lines = body.split("\n")
|
||
tenant_i = None
|
||
for i, ln in enumerate(lines):
|
||
if TENANT_COL_RX.match(ln):
|
||
tenant_i = i
|
||
break
|
||
if tenant_i is None:
|
||
skips.append((tbl, line_of(sql, m.start()), "未找到 tenant_id 列定义，跳过"))
|
||
continue
|
||
first_meaningful = None
|
||
for i, ln in enumerate(lines):
|
||
if ln.strip():
|
||
first_meaningful = i
|
||
break
|
||
if first_meaningful is None or tenant_i == first_meaningful:
|
||
continue # 已是首列，幂等无操作
|
||
moved = lines.pop(tenant_i)
|
||
lines.insert(first_meaningful, moved)
|
||
new_body = "\n".join(lines)
|
||
start = open_idx + 1 + offset_shift
|
||
end = close_idx + offset_shift
|
||
out = out[:start] + new_body + out[end:]
|
||
offset_shift += len(new_body) - len(body)
|
||
changes.append((tbl, line_of(sql, m.start()),
|
||
"tenant_id 由第 %d 个列定义提升为物理首列" % (tenant_i - first_meaningful + 1)))
|
||
return out, changes, skips
|
||
|
||
|
||
def main():
|
||
ap = argparse.ArgumentParser(description="DDL tenant_id 物理首列幂等修正器")
|
||
ap.add_argument("--file", required=True, help="DDL 文件路径")
|
||
ap.add_argument("--dry-run", action="store_true", help="只报告不落盘")
|
||
args = ap.parse_args()
|
||
|
||
if not os.path.isfile(args.file):
|
||
sys.stderr.write("[patch_ddl] 文件不存在：%s\n" % args.file)
|
||
return 2
|
||
with io.open(args.file, "r", encoding="utf-8") as fh:
|
||
sql = fh.read()
|
||
|
||
new_sql, changes, skips = patch(sql)
|
||
for tbl, ln, msg in changes:
|
||
print("[PATCH] %-34s CREATE @L%-5d %s" % (tbl, ln, msg))
|
||
for tbl, ln, msg in skips:
|
||
print("[SKIP ] %-34s CREATE @L%-5d %s" % (tbl, ln, msg))
|
||
print("[SUMMARY] 变更 %d 张表，跳过 %d 张表，幂等=%s"
|
||
% (len(changes), len(skips), "是（重复执行不再变更）" if not changes else "本次有变更"))
|
||
|
||
if changes and not args.dry_run:
|
||
with io.open(args.file, "w", encoding="utf-8") as fh:
|
||
fh.write(new_sql)
|
||
print("[WRITE] 已落盘：%s（%d 字节）" % (args.file, os.path.getsize(args.file)))
|
||
elif args.dry_run:
|
||
print("[DRY-RUN] 未落盘")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|