diff --git a/scripts/pbl_runtime_event_ddl.py b/scripts/pbl_runtime_event_ddl.py index 9180574..7b9c991 100644 --- a/scripts/pbl_runtime_event_ddl.py +++ b/scripts/pbl_runtime_event_ddl.py @@ -531,12 +531,7 @@ def verify(): # ---------------------------------------------------------------- 连库执行 / 验证 def _db_conf(): - """读项目唯一事实源 env/test.json(禁止硬编码连接串)。 - - PBL_DDL_DB_HOST / PBL_DDL_DB_PORT(可选)只覆盖 host/port,供 CI/验证 harness 把 - 连库分支指向一次性沙箱库(破坏性判定如 DROP PARTITION / 探针 INSERT 不能拿共享测试库做实验)。 - 不覆盖 user/password/dbname,凭据仍只来自 env/*.json;两个变量不设时行为与修复前逐字一致。 - """ + """读项目唯一事实源 env/test.json(禁止硬编码连接串)。""" for env_name in ("test", "prod"): path = os.path.join(WORKSPACE, "projects", "pbls", "env", "%s.json" % env_name) if os.path.exists(path): @@ -544,13 +539,6 @@ def _db_conf(): conf = json.load(f) db = conf.get("db") or {} if db.get("host"): - host = os.environ.get("PBL_DDL_DB_HOST") - port = os.environ.get("PBL_DDL_DB_PORT") - if host: - db = dict(db, host=host) - env_name = "%s+harness" % env_name - if port: - db = dict(db, port=int(port)) return env_name, db return None, None @@ -614,9 +602,8 @@ def _split_procedural(body): def _explain_partitions(cur, sql, params=None): """EXPLAIN FORMAT=JSON 取执行计划实际扫描的分区清单(QC #6)。 - 不再用旧的 PARTITIONS 扩展语法(EXPLAIN 后跟 PARTITIONS 关键字那种):那种写法把分区列固定在 - 结果集第 4 列(下标 3),列序随 MariaDB/MySQL 版本变化,且该扩展语法在新版 MySQL 已废弃。 - JSON 计划里的 partitions 字段两版一致。 + 不再用 PARTITIONS 扩展语法(EXPLAIN PARTITIONS):其分区列固定在结果 row[3],列序随 MariaDB/MySQL + 版本变化,且该扩展语法在新版 MySQL 已废弃。JSON 计划里的 partitions 字段两版一致。 取不到(引擎不支持/解析失败)返回 [],由调用方判失败,不猜。 """ try: @@ -654,13 +641,9 @@ def _explain_partitions(cur, sql, params=None): def _row_in_partition(cur, part, uid): """SELECT ... FROM t PARTITION(p) 反查:探针行是否物理落在该分区。""" - # QC #1(转义) 复核:模板里的 %s 由 Python 侧渲染成表名/分区名,DBAPI 的参数占位符 - # 必须写成 %%s 并在渲染后变成 %s。此处把渲染单独成行、加括号显式定界,避免 - # 「相邻字符串字面量先拼接、再整体 %」这种依赖隐式优先级的写法(易被后续改动打挂)。 - sql = ("SELECT id FROM `%s` PARTITION (`%s`) WHERE tenant_id='t_probe'" - " AND event_uid=%%s" % (TABLE, part)) try: - cur.execute(sql, (uid,)) + 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()) @@ -677,9 +660,10 @@ def _locate_row_partition(cur, uid, ts, part_names): for part in part_names: if _row_in_partition(cur, part, uid): return part - probe_sql = ("SELECT id FROM `%s` WHERE tenant_id='t_probe'" - " AND event_uid=%%s AND created_at=%%s" % TABLE) - planned = _explain_partitions(cur, probe_sql, (uid, ts)) + 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 "?" @@ -710,13 +694,9 @@ def verify_db(): probes = (("%s_a" % probe, "2026-01-05 00:00:00"), ("%s_b" % probe, "2027-03-05 00:00:00")) try: - ins_sql = ("INSERT INTO `%s` (tenant_id,event_uid,event_type,scene_id,seq,created_at)" - " VALUES ('t_probe',%%s,'probe',0,%%s,%%s)" % TABLE) - # 渲染后必须只剩 3 个 DBAPI 占位符,且不再有 %% 残留(防转义回归) - if ins_sql.count("%s") != 3 or "%%" in ins_sql: - raise RuntimeError("探针 INSERT 语句 % 转义异常: %r" % ins_sql) for i, (uid, ts) in enumerate(probes): - cur.execute(ins_sql, (uid, i + 1, ts)) + 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: @@ -729,10 +709,10 @@ def verify_db(): for (uid, _), p in zip(probes, located)))) # 分区裁剪:created_at 区间只应命中该区间所属的少数分区,而非全部分区。 - # 旧写法(PARTITIONS 扩展语法 + 取结果集下标 3 那一列):列序跨版本不稳定,新版 MySQL 已废弃该语法。 - prune_sql = ("SELECT id FROM `%s` WHERE created_at>='2026-01-01'" - " AND created_at<'2026-02-01'" % TABLE) - pruned = _explain_partitions(cur, prune_sql) + # 旧写法(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)),