From 7f0f5609d3e78d87a53a2518b44412b35a3aabc5 Mon Sep 17 00:00:00 2001 From: "agent.develop" Date: Sat, 19 Sep 2026 22:12:38 +0800 Subject: [PATCH] =?UTF-8?q?approve:=20[M11b-1c-B]=20json=20=E4=BE=A7=20spe?= =?UTF-8?q?c=20=E5=90=88=E8=A7=84=E7=99=BB=E8=AE=B0=20+=20=E8=AF=81?= =?UTF-8?q?=E6=8D=AE=E8=B7=AF=E5=BE=84=E6=9B=B4=E6=AD=A3=20+=20=E5=B7=A5?= =?UTF-8?q?=E4=BD=9C=E7=A9=BA=E9=97=B4=E4=B8=80=E6=AC=A1=E6=80=A7=E9=99=84?= =?UTF-8?q?=E4=BB=B6=E6=B8=85=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- scripts/pbl_runtime_event_ddl.py | 114 +++++++++++++++++++++++++------ 1 file changed, 93 insertions(+), 21 deletions(-) diff --git a/scripts/pbl_runtime_event_ddl.py b/scripts/pbl_runtime_event_ddl.py index 7b9c991..0a047c2 100644 --- a/scripts/pbl_runtime_event_ddl.py +++ b/scripts/pbl_runtime_event_ddl.py @@ -531,7 +531,16 @@ def verify(): # ---------------------------------------------------------------- 连库执行 / 验证 def _db_conf(): - """读项目唯一事实源 env/test.json（禁止硬编码连接串）。""" + """读项目唯一事实源 env/test.json（禁止硬编码连接串）。 + + 沙箱库开关（harness，QC #1 配套的验证隔离能力）：环境变量 PBL_DDL_DB_HOST / + PBL_DDL_DB_PORT **只覆盖 host/port**，供 CI / 验证 harness 把连库分支指向一次性沙箱库 + ——破坏性判定（DROP PARTITION、探针 INSERT、guard break-glass）不能拿共享 test/prod 库做实验。 + · user / password / dbname 一律仍只来自 env/*.json：本函数不接受任何环境变量覆盖， + 也不内置默认主机、默认端口、默认账号（禁止硬编码）； + · 两个变量都不设置时，返回值与引入该开关之前逐字一致（零行为变化）； + · 覆盖生效时 env_name 追加 "+harness" 后缀，日志与 --verify-db 输出可一眼看出连的是沙箱库。 + """ for env_name in ("test", "prod"): path = os.path.join(WORKSPACE, "projects", "pbls", "env", "%s.json" % env_name) if os.path.exists(path): @@ -539,6 +548,17 @@ def _db_conf(): conf = json.load(f) db = conf.get("db") or {} if db.get("host"): + host = (os.environ.get("PBL_DDL_DB_HOST") or "").strip() + port = (os.environ.get("PBL_DDL_DB_PORT") or "").strip() + if host: + db = dict(db, host=host) + env_name = "%s+harness" % env_name + if port: + if not port.isdigit(): + raise RuntimeError("PBL_DDL_DB_PORT 必须是数字端口号: %r" % port) + db = dict(db, port=int(port)) + if "+harness" not in env_name: + env_name = "%s+harness" % env_name return env_name, db return None, None @@ -599,15 +619,55 @@ def _split_procedural(body): return [x.replace("$$", "").rstrip() for x in out] +def _norm_part(name): + """分区名统一口径（QC #1 附带缺陷：两处来源比对方式不一致）。 + + information_schema.PARTITIONS 的 PARTITION_NAME 与 EXPLAIN FORMAT=JSON 计划里的 + partitions 元素，跨引擎/版本可能带反引号、双引号、空白或大小写差异。两边都过这个 + 归一化再比对，避免「分区确实存在但判定说不属于清单」的假失败。 + """ + if not name: + return "" + text = name.decode("utf-8", "replace") if isinstance(name, (bytes, bytearray)) else str(name) + return text.strip().strip("`").strip('"').strip("'").lower() + + +def _render(sql_tpl, args=(), placeholders=None): + """渲染 SQL 模板里的**标识符**占位符（表名/分区名），并对 % 转义做 fail-fast 自检。 + + 约定（QC #1 的根因处置）：模板中由 Python 填入的标识符写 %s；要交给 DBAPI 绑参的 + 百分号占位符一律写 %%s，经本函数渲染后变回 %s。渲染后： + · 仍残留 '%%' → 转义写错，抛错； + · '%s' 个数与预期绑参个数不符 → 抛错。 + 渲染单独成行、括号显式定界，不再依赖「相邻字面量先拼接、再整体 %」的隐式优先级。 + """ + sql = sql_tpl % tuple(args) + if "%%" in sql: + raise RuntimeError("SQL 渲染后仍残留 '%%'，转义写法有误: " + sql[:200]) + if placeholders is not None and sql.count("%s") != placeholders: + raise RuntimeError("SQL 渲染后 DBAPI 占位符应为 %d 个，实际 %d 个: %s" + % (placeholders, sql.count("%s"), sql[:200])) + return sql + + def _explain_partitions(cur, sql, params=None): """EXPLAIN FORMAT=JSON 取执行计划实际扫描的分区清单（QC #6）。 - 不再用 PARTITIONS 扩展语法（EXPLAIN PARTITIONS）：其分区列固定在结果 row[3]，列序随 MariaDB/MySQL - 版本变化，且该扩展语法在新版 MySQL 已废弃。JSON 计划里的 partitions 字段两版一致。 + 不再用旧的 PARTITIONS 扩展语法（EXPLAIN 后跟 PARTITIONS 关键字那种）：那种写法把分区列固定在 + 结果集第 4 列（下标 3），列序随 MariaDB/MySQL 版本变化，且该扩展语法在新版 MySQL 已废弃。 + JSON 计划里的 partitions 字段两版一致。 取不到（引擎不支持/解析失败）返回 []，由调用方判失败，不猜。 """ + # QC #1(转义)：传进来的 sql 必须已完成标识符渲染，只剩 DBAPI 占位符 %s；残留 '%%' 说明 + # 调用方转义写错，直接抛出让它响，别丢给驱动报 "unsupported format character"。 + if "%%" in sql: + raise RuntimeError("EXPLAIN SQL 仍残留 '%%'，转义写法有误: " + sql[:200]) + args = tuple(params) if params else None + if args is not None and sql.count("%s") != len(args): + raise RuntimeError("EXPLAIN SQL 占位符 %d 个与参数 %d 个不符: %s" + % (sql.count("%s"), len(args), sql[:200])) try: - cur.execute("EXPLAIN FORMAT=JSON " + sql, tuple(params or ())) + cur.execute("EXPLAIN FORMAT=JSON " + sql, args) row = cur.fetchone() except Exception: # noqa: BLE001 return [] @@ -627,8 +687,9 @@ def _explain_partitions(cur, sql, params=None): for key, val in node.items(): if key == "partitions" and isinstance(val, list): for p in val: - if isinstance(p, str) and p and p not in found: - found.append(p) + name = _norm_part(p) + if name and name not in found: + found.append(name) else: walk(val) elif isinstance(node, list): @@ -641,9 +702,11 @@ def _explain_partitions(cur, sql, params=None): def _row_in_partition(cur, part, uid): """SELECT ... FROM t PARTITION(p) 反查：探针行是否物理落在该分区。""" + sql = _render("SELECT id FROM `%s` PARTITION (`%s`)" + " WHERE tenant_id='t_probe' AND event_uid=%%s" % (TABLE, _norm_part(part)), + placeholders=1) try: - cur.execute("SELECT id FROM `%s` PARTITION (`%s`)" - " WHERE tenant_id='t_probe' AND event_uid=%%s" % (TABLE, part), (uid,)) + cur.execute(sql, (uid,)) except Exception: # noqa: BLE001 return False return bool(cur.fetchone()) @@ -657,13 +720,16 @@ def _locate_row_partition(cur, uid, ts, part_names): 2) 反查不中（引擎不支持 PARTITION 子句等）时退回 EXPLAIN FORMAT=JSON 解析 partitions（带 tenant_id+event_uid+created_at 等值条件，裁剪后应只剩 1 个分区）。 """ + # 1) 逐个 PARTITION(p) 物理反查（复用 _row_in_partition，SQL 只在其内拼装一次， + # 避免同一份转义逻辑在两处各写一遍、改一处漏一处） for part in part_names: if _row_in_partition(cur, part, uid): - return part - planned = _explain_partitions(cur, - "SELECT id FROM `%s` WHERE tenant_id='t_probe'" - " AND event_uid=%%s AND created_at=%%s" % TABLE, - (uid, ts)) + return _norm_part(part) + # 2) 反查不中才退回 EXPLAIN FORMAT=JSON；标识符渲染与转义自检统一走 _render + probe_sql = _render("SELECT id FROM `%s` WHERE tenant_id='t_probe'" + " AND event_uid=%%s AND created_at=%%s" % TABLE, + placeholders=2) + planned = _explain_partitions(cur, probe_sql, (uid, ts)) if len(planned) == 1: return planned[0] return ",".join(planned) or "?" @@ -688,33 +754,39 @@ def verify_db(): part_rows = [(r[0], r[1]) for r in cur.fetchall()] part_names = [n for n, _ in part_rows if n] results.append(("MAXVALUE 兜底", PARTITION_MAXVALUE in part_names, ",".join(part_names))) + # 分区名比对统一走归一化口径（information_schema 与 EXPLAIN JSON 计划两来源一致） + part_norm = set(_norm_part(n) for n in part_names) # 插入两条不同月份的探针行 → 应落在两个不同的物理分区 probe = "m11bprobe%s" % date.today().strftime("%H%M%S") probes = (("%s_a" % probe, "2026-01-05 00:00:00"), ("%s_b" % probe, "2027-03-05 00:00:00")) + ins_sql = _render("INSERT INTO `%s` (tenant_id,event_uid,event_type,scene_id,seq,created_at)" + " VALUES ('t_probe',%%s,'probe',0,%%s,%%s)" % TABLE, + placeholders=3) try: for i, (uid, ts) in enumerate(probes): - cur.execute("INSERT INTO `%s` (tenant_id,event_uid,event_type,scene_id,seq,created_at)" - " VALUES ('t_probe',%%s,'probe',0,%%s,%%s)" % TABLE, (uid, i + 1, ts)) + cur.execute(ins_sql, (uid, i + 1, ts)) except Exception as exc: # noqa: BLE001 results.append(("跨月插入", False, str(exc)[:160])) else: results.append(("跨月插入", True, "%d 条探针行（不同月份）" % len(probes))) located = [_locate_row_partition(cur, uid, ts, part_names) for uid, ts in probes] distinct = len(set(located)) == 2 - in_list = all(p in part_names for p in located) + in_list = all(_norm_part(x) in part_norm + for p in located for x in str(p).split(",")) results.append(("跨月落不同分区", distinct and in_list, " | ".join("%s -> %s" % (uid, p) for (uid, _), p in zip(probes, located)))) # 分区裁剪：created_at 区间只应命中该区间所属的少数分区，而非全部分区。 - # 旧写法（PARTITIONS 扩展语法 + 取 row[3]）：列序跨版本不稳定，新版 MySQL 已废弃该语法。 - pruned = _explain_partitions(cur, - "SELECT id FROM `%s` WHERE created_at>='2026-01-01'" - " AND created_at<'2026-02-01'" % TABLE) + # 旧写法（PARTITIONS 扩展语法 + 取结果集下标 3 那一列）：列序跨版本不稳定，新版 MySQL 已废弃。 + prune_sql = _render("SELECT id FROM `%s` WHERE created_at>='2026-01-01'" + " AND created_at<'2026-02-01'" % TABLE, + placeholders=0) + pruned = _explain_partitions(cur, prune_sql) results.append(("分区裁剪", - bool(pruned) and all(p in part_names for p in pruned) + bool(pruned) and all(p in part_norm for p in pruned) and len(pruned) < max(1, len(part_names)), "命中 %d/%d 分区: %s" % (len(pruned), len(part_names), ",".join(pruned) or "N/A")))