From 049e09af376088712d235e0f1c9fb6c8cbbdcaee Mon Sep 17 00:00:00 2001 From: "agent.develop" Date: Sat, 19 Sep 2026 21:21:04 +0800 Subject: [PATCH] =?UTF-8?q?deliver:=20=E4=BA=A4=E4=BB=98=E6=94=B6=E5=8F=A3?= =?UTF-8?q?=EF=BC=88=E5=BC=95=E6=93=8E=E4=BB=A3=E4=B8=BA=E6=8F=90=E4=BA=A4?= =?UTF-8?q?=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- scripts/_m11b1a_patch.py | 177 +++++++++++++++++++++++++++++++ scripts/pbl_runtime_event_ddl.py | 165 ++++++++++++++++++++-------- 2 files changed, 299 insertions(+), 43 deletions(-) create mode 100644 scripts/_m11b1a_patch.py diff --git a/scripts/_m11b1a_patch.py b/scripts/_m11b1a_patch.py new file mode 100644 index 0000000..608ef90 --- /dev/null +++ b/scripts/_m11b1a_patch.py @@ -0,0 +1,177 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +# M11b-1a 定点补丁脚本（一次性工具，跑完即删）： +# Patch1 maintenance_sql(): 百分号格式化渲染 → token + str.replace，避开 % 格式化崩溃（QC #1） +# Patch2 verify_db(): 删空转查询；落分区/分区裁剪改用 information_schema 分区清单 +# + SELECT ... PARTITION(p) 反查 + EXPLAIN FORMAT=JSON 解析 partitions（QC #6） +# Patch3 新增连库辅助函数 _explain_partitions/_row_in_partition/_locate_row_partition +# DDL 列定义/索引/分区计划文本一律不动。 +import os + +HERE = os.path.dirname(os.path.abspath(__file__)) +P = os.path.join(os.path.dirname(HERE), "scripts", "pbl_runtime_event_ddl.py") +src = open(P, encoding="utf-8").read() +orig = src + +# ================================================================ Patch 1 +i0 = src.index("def maintenance_sql(") +i1 = src.index("def migrate_from_legacy_sql()") +region = src[i0:i1] +assert "%(table)s" in region and "%(pmax)s" in region and "%(months)s" in region, "占位符缺失" + +new_region = (region + .replace("%(table)s", "__TABLE__") + .replace("%(pmax)s", "__PMAX__") + .replace("%(months)s", "__MONTHS__")) + +old_doc = ' """分区维护：过程（幂等补建月分区 / 到期 DROP 旧分区）+ 每月 EVENT。"""' +new_doc = (' """分区维护：过程（幂等补建月分区 / 到期 DROP 旧分区）+ 每月 EVENT。\n' + '\n' + " QC #1 修复：正文含大量字面百分号（DATE_FORMAT(...,'%Y-%m-01')、'%Y%m'），\n" + ' 原先用「三引号 % 字典」渲染会抛 ValueError: unsupported format character。\n' + ' 现改用 __TABLE__/__PMAX__/__MONTHS__ token + str.replace，\n' + ' 字面百分号原样输出、无需再写双写转义。\n' + ' """') +assert old_doc in new_region, "docstring 未命中" +new_region = new_region.replace(old_doc, new_doc) + +old_tail = ('""" % {"table": TABLE, "pmax": PARTITION_MAXVALUE, "months": months,\n' + ' "today": (today or date.today()).isoformat()}') +new_tail = ('""".replace("__TABLE__", TABLE)\n' + ' .replace("__PMAX__", PARTITION_MAXVALUE)\n' + ' .replace("__MONTHS__", str(months))') +assert old_tail in new_region, "maintenance_sql 渲染尾段未命中" +new_region = new_region.replace(old_tail, new_tail) + +src = src[:i0] + new_region + src[i1:] + +# ================================================================ Patch 2 +s2 = src.index(' cur.execute("SELECT PARTITION_NAME FROM information_schema.PARTITIONS "') +e2_marker = (' results.append(("分区裁剪", pruned.count(",") == 0 and "p" in pruned,' + ' "EXPLAIN partitions=%s" % pruned))') +e2 = src.index(e2_marker) + len(e2_marker) +old_block = src[s2:e2] +assert "EXPLAIN PARTITIONS SELECT id FROM `%s` WHERE event_uid" in old_block, "旧判定块未命中" +assert "for k in" in old_block, "空转 for 循环不在预期块内" + +new_block = ''' # 分区清单：以 information_schema.PARTITIONS 为权威来源（按 ordinal 排序）， + # 后面「落哪个分区」「分区裁剪」两项判定都基于这份清单（QC #6） + cur.execute("SELECT PARTITION_NAME, PARTITION_DESCRIPTION FROM information_schema.PARTITIONS " + "WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=%s " + "AND PARTITION_NAME IS NOT NULL ORDER BY PARTITION_ORDINAL_POSITION", (TABLE,)) + 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))) + + # 插入两条不同月份的探针行 → 应落在两个不同的物理分区 + 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")) + 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)) + 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) + results.append(("跨月落不同分区", distinct and in_list, + " | ".join("%s -> %s" % (uid, p) + for (uid, _), p in zip(probes, located)))) + + # 分区裁剪：created_at 区间只应命中该区间所属的少数分区，而非全部分区。 + # 旧写法 EXPLAIN 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) + results.append(("分区裁剪", + bool(pruned) and all(p in part_names 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")))''' + +src = src[:s2] + new_block + src[e2:] + +# ================================================================ Patch 3 +helpers = '''def _explain_partitions(cur, sql, params=None): + """EXPLAIN FORMAT=JSON 取执行计划实际扫描的分区清单（QC #6）。 + + 不再用 EXPLAIN PARTITIONS：其分区列固定在结果 row[3]，列序随 MariaDB/MySQL + 版本变化，且该扩展语法在新版 MySQL 已废弃。JSON 计划里的 partitions 字段两版一致。 + 取不到（引擎不支持/解析失败）返回 []，由调用方判失败，不猜。 + """ + try: + cur.execute("EXPLAIN FORMAT=JSON " + sql, tuple(params or ())) + row = cur.fetchone() + except Exception: # noqa: BLE001 + return [] + raw = row[0] if row else None + if isinstance(raw, (bytes, bytearray)): + raw = raw.decode("utf-8", "replace") + if not raw: + return [] + try: + plan = json.loads(raw) + except (ValueError, TypeError): + return [] + found = [] + + def walk(node): + if isinstance(node, dict): + 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) + else: + walk(val) + elif isinstance(node, list): + for val in node: + walk(val) + + walk(plan) + return found + + +def _row_in_partition(cur, part, uid): + """SELECT ... FROM t PARTITION(p) 反查：探针行是否物理落在该分区。""" + try: + cur.execute("SELECT id FROM `%s` PARTITION (`%s`)" + " WHERE tenant_id='t_probe' AND event_uid=%%s" % (TABLE, part), (uid,)) + except Exception: # noqa: BLE001 + return False + return bool(cur.fetchone()) + + +def _locate_row_partition(cur, uid, ts, part_names): + """定位某探针行实际所在分区（QC #6）。 + + 1) 首选按 information_schema 分区清单逐个 PARTITION(p) 反查物理位置——与 EXPLAIN + 输出列序无关，最可靠； + 2) 反查不中（引擎不支持 PARTITION 子句等）时退回 EXPLAIN FORMAT=JSON 解析 + partitions（带 tenant_id+event_uid+created_at 等值条件，裁剪后应只剩 1 个分区）。 + """ + 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)) + if len(planned) == 1: + return planned[0] + return ",".join(planned) or "?" + + +''' +marker = "def verify_db():" +assert src.count(marker) == 1, "verify_db 定位异常" +src = src.replace(marker, helpers + marker) + +open(P, "w", encoding="utf-8").write(src) +print("PATCHED %s: %d -> %d bytes" % (P, len(orig), len(src))) +print("residual 'EXPLAIN PARTITIONS ' count:", src.count("EXPLAIN PARTITIONS ")) +print("residual 'for k in' count:", src.count("for k in")) diff --git a/scripts/pbl_runtime_event_ddl.py b/scripts/pbl_runtime_event_ddl.py index b7472a4..7b9c991 100644 --- a/scripts/pbl_runtime_event_ddl.py +++ b/scripts/pbl_runtime_event_ddl.py @@ -243,9 +243,15 @@ END; def maintenance_sql(today=None, months=None): - """分区维护：过程（幂等补建月分区 / 到期 DROP 旧分区）+ 每月 EVENT。""" + """分区维护：过程（幂等补建月分区 / 到期 DROP 旧分区）+ 每月 EVENT。 + + QC #1 修复：正文含大量字面百分号（DATE_FORMAT(...,'%Y-%m-01')、'%Y%m'）， + 原先用「三引号 % 字典」渲染会抛 ValueError: unsupported format character。 + 现改用 __TABLE__/__PMAX__/__MONTHS__ token + str.replace， + 字面百分号原样输出、无需再写双写转义。 + """ months = int(months or MONTHS_PRECREATE) - return """-- 1) 幂等补建未来月分区：pmax 存在时用 REORGANIZE PARTITION 拆出新月份（不丢数据） + body = """-- 1) 幂等补建未来月分区：pmax 存在时用 REORGANIZE PARTITION 拆出新月份（不丢数据） DROP PROCEDURE IF EXISTS `pbl_runtime_event_ensure_partitions`; CREATE PROCEDURE `pbl_runtime_event_ensure_partitions`(IN p_months_ahead INT) MODIFIES SQL DATA @@ -274,13 +280,13 @@ BEGIN LEAVE read_loop; END IF; SELECT COUNT(*) INTO v_exists FROM information_schema.PARTITIONS - WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = '%(table)s' + WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = '__TABLE__' AND PARTITION_NAME = v_name; IF v_exists = 0 THEN SET @pbl_part_sql = CONCAT( - 'ALTER TABLE `%(table)s` REORGANIZE PARTITION `%(pmax)s` INTO (', + 'ALTER TABLE `__TABLE__` REORGANIZE PARTITION `__PMAX__` INTO (', 'PARTITION `', v_name, '` VALUES LESS THAN (DATE ''', v_upper, '''), ', - 'PARTITION `%(pmax)s` VALUES LESS THAN (MAXVALUE))'); + 'PARTITION `__PMAX__` VALUES LESS THAN (MAXVALUE))'); PREPARE pbl_part_st FROM @pbl_part_sql; EXECUTE pbl_part_st; DEALLOCATE PREPARE pbl_part_st; @@ -299,9 +305,9 @@ BEGIN DECLARE v_cutoff DATE; DECLARE cur CURSOR FOR SELECT PARTITION_NAME FROM information_schema.PARTITIONS - WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = '%(table)s' + WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = '__TABLE__' AND PARTITION_NAME IS NOT NULL - AND PARTITION_NAME <> '%(pmax)s' + AND PARTITION_NAME <> '__PMAX__' AND PARTITION_NAME < CONCAT('p', DATE_FORMAT(v_cutoff, '%Y%m')); DECLARE CONTINUE HANDLER FOR NOT FOUND SET v_done = 1; SET v_cutoff = DATE_ADD(DATE_FORMAT(CURDATE(), '%Y-%m-01'), INTERVAL -p_keep_months MONTH); @@ -311,7 +317,7 @@ BEGIN IF v_done = 1 THEN LEAVE prune_loop; END IF; - SET @pbl_drop_sql = CONCAT('ALTER TABLE `%(table)s` DROP PARTITION `', v_name, '`'); + SET @pbl_drop_sql = CONCAT('ALTER TABLE `__TABLE__` DROP PARTITION `', v_name, '`'); PREPARE pbl_drop_st FROM @pbl_drop_sql; EXECUTE pbl_drop_st; DEALLOCATE PREPARE pbl_drop_st; @@ -321,16 +327,19 @@ END; -- 3) 每月自动补分区（需 event_scheduler=ON；否则用 cron，见开发说明） -- SET GLOBAL event_scheduler = ON; -DROP EVENT IF EXISTS `ev_%(table)s_partition_maintain`; -CREATE EVENT IF NOT EXISTS `ev_%(table)s_partition_maintain` +DROP EVENT IF EXISTS `ev___TABLE___partition_maintain`; +CREATE EVENT IF NOT EXISTS `ev___TABLE___partition_maintain` ON SCHEDULE EVERY 1 MONTH STARTS TIMESTAMP(DATE_FORMAT(DATE_ADD(CURDATE(), INTERVAL 1 MONTH), '%Y-%m-01 03:00:00')) ON COMPLETION PRESERVE ENABLE - COMMENT '每月预建未来 %(months)s 个月分区' -DO CALL pbl_runtime_event_ensure_partitions(%(months)s); -""" % {"table": TABLE, "pmax": PARTITION_MAXVALUE, "months": months, - "today": (today or date.today()).isoformat()} + COMMENT '每月预建未来 __MONTHS__ 个月分区' +DO CALL pbl_runtime_event_ensure_partitions(__MONTHS__); +""" + return (body + .replace("__TABLE__", TABLE) + .replace("__PMAX__", PARTITION_MAXVALUE) + .replace("__MONTHS__", str(months))) def migrate_from_legacy_sql(): @@ -590,6 +599,76 @@ def _split_procedural(body): return [x.replace("$$", "").rstrip() for x in out] +def _explain_partitions(cur, sql, params=None): + """EXPLAIN FORMAT=JSON 取执行计划实际扫描的分区清单（QC #6）。 + + 不再用 PARTITIONS 扩展语法（EXPLAIN PARTITIONS）：其分区列固定在结果 row[3]，列序随 MariaDB/MySQL + 版本变化，且该扩展语法在新版 MySQL 已废弃。JSON 计划里的 partitions 字段两版一致。 + 取不到（引擎不支持/解析失败）返回 []，由调用方判失败，不猜。 + """ + try: + cur.execute("EXPLAIN FORMAT=JSON " + sql, tuple(params or ())) + row = cur.fetchone() + except Exception: # noqa: BLE001 + return [] + raw = row[0] if row else None + if isinstance(raw, (bytes, bytearray)): + raw = raw.decode("utf-8", "replace") + if not raw: + return [] + try: + plan = json.loads(raw) + except (ValueError, TypeError): + return [] + found = [] + + def walk(node): + if isinstance(node, dict): + 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) + else: + walk(val) + elif isinstance(node, list): + for val in node: + walk(val) + + walk(plan) + return found + + +def _row_in_partition(cur, part, uid): + """SELECT ... FROM t PARTITION(p) 反查：探针行是否物理落在该分区。""" + try: + cur.execute("SELECT id FROM `%s` PARTITION (`%s`)" + " WHERE tenant_id='t_probe' AND event_uid=%%s" % (TABLE, part), (uid,)) + except Exception: # noqa: BLE001 + return False + return bool(cur.fetchone()) + + +def _locate_row_partition(cur, uid, ts, part_names): + """定位某探针行实际所在分区（QC #6）。 + + 1) 首选按 information_schema 分区清单逐个 PARTITION(p) 反查物理位置——与 EXPLAIN + 输出列序无关，最可靠； + 2) 反查不中（引擎不支持 PARTITION 子句等）时退回 EXPLAIN FORMAT=JSON 解析 + partitions（带 tenant_id+event_uid+created_at 等值条件，裁剪后应只剩 1 个分区）。 + """ + 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)) + if len(planned) == 1: + return planned[0] + return ",".join(planned) or "?" + + def verify_db(): """连库验证：分区裁剪 / 数据落分区 / UPDATE·DELETE 被拒。""" conn, env_name, db = _connect() @@ -601,44 +680,44 @@ def verify_db(): n_parts = cur.fetchone()[0] results.append(("分区数", n_parts >= MONTHS_PRECREATE + 1, "实际 %d（含 pmax）" % n_parts)) - cur.execute("SELECT PARTITION_NAME FROM information_schema.PARTITIONS " + # 分区清单：以 information_schema.PARTITIONS 为权威来源（按 ordinal 排序）， + # 后面「落哪个分区」「分区裁剪」两项判定都基于这份清单（QC #6） + cur.execute("SELECT PARTITION_NAME, PARTITION_DESCRIPTION FROM information_schema.PARTITIONS " "WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=%s " "AND PARTITION_NAME IS NOT NULL ORDER BY PARTITION_ORDINAL_POSITION", (TABLE,)) - names = [r[0] for r in cur.fetchall()] - results.append(("MAXVALUE 兜底", PARTITION_MAXVALUE in names, ",".join(names))) + 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))) - # 插入两个不同月份 → 落不同分区 + # 插入两条不同月份的探针行 → 应落在两个不同的物理分区 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")) try: - cur.execute("INSERT INTO `%s` (tenant_id,event_uid,event_type,scene_id,seq,created_at)" - " VALUES ('t_probe','%s_a','probe',0,1,'2026-01-05 00:00:00')" % (TABLE, probe)) - cur.execute("INSERT INTO `%s` (tenant_id,event_uid,event_type,scene_id,seq,created_at)" - " VALUES ('t_probe','%s_b','probe',0,2,'2027-03-05 00:00:00')" % (TABLE, probe)) + 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)) except Exception as exc: # noqa: BLE001 results.append(("跨月插入", False, str(exc)[:160])) else: - cur.execute("SELECT PARTITION_NAME FROM (SELECT '%s_a' k" - " UNION ALL SELECT '%s_b') x" % (probe, probe)) - parts = set() - for k in ("%s_a" % probe, "%s_b" % probe): - cur.execute("SELECT PARTITION_NAME FROM information_schema.PARTITIONS" - " WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=%s" - " AND PARTITION_DESCRIPTION <> 'MAXVALUE'" - " AND PARTITION_DESCRIPTION IS NOT NULL", (TABLE,)) - # 精确定位：EXPLAIN 每行 - located = [] - for uid in ("%s_a" % probe, "%s_b" % probe): - cur.execute("EXPLAIN PARTITIONS SELECT id FROM `%s` WHERE event_uid=%%s" % TABLE, (uid,)) - row = cur.fetchone() - located.append(str(row[3] if row and len(row) > 3 else row)) - results.append(("跨月落不同分区", len(set(located)) == 2, " | ".join(located))) + 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) + results.append(("跨月落不同分区", distinct and in_list, + " | ".join("%s -> %s" % (uid, p) + for (uid, _), p in zip(probes, located)))) - # EXPLAIN 分区裁剪 - cur.execute("EXPLAIN PARTITIONS SELECT id FROM `%s` WHERE created_at>='2026-01-01'" - " AND created_at<'2026-02-01'" % TABLE) - row = cur.fetchone() - pruned = str(row[3] if row and len(row) > 3 else row) - results.append(("分区裁剪", pruned.count(",") == 0 and "p" in pruned, "EXPLAIN partitions=%s" % pruned)) + # 分区裁剪：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) + results.append(("分区裁剪", + bool(pruned) and all(p in part_names 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"))) # append-only：UPDATE / DELETE 必须被拒 for verb, sql in (("UPDATE", "UPDATE `%s` SET event_type='x' WHERE tenant_id='t_probe'" % TABLE),