revert(M11b-1c): 还原越界生成的 scripts/pbl_runtime_event_ddl.py 至基线 4a7c7b9(blob 7b9c991)

M11b-1c 任务书边界=不改生成器、DDL 侧为权威派生源；309d374 与引擎代提交 7e7477f
一并改动了 scripts/pbl_runtime_event_ddl.py（_row_in_partition/_locate_row_partition/
verify_db 的 % 转义重构 + PBL_DDL_DB_HOST/PORT harness 覆盖），触发 QC #1/#2 越界硬门禁。
本提交仅还原该 .py，models/pbl_runtime_event.json 本体一字未动。

scripts/pbl_runtime_event_ddl.py

@@ -531,12 +531,7 @@ def verify():
 
 # ---------------------------------------------------------------- 连库执行 / 验证
 def _db_conf():
-    """读项目唯一事实源 env/test.json（禁止硬编码连接串）。
+    """读项目唯一事实源 env/test.json（禁止硬编码连接串）。"""
-
-    PBL_DDL_DB_HOST / PBL_DDL_DB_PORT（可选）只覆盖 host/port，供 CI/验证 harness 把
-    连库分支指向一次性沙箱库（破坏性判定如 DROP PARTITION / 探针 INSERT 不能拿共享测试库做实验）。
-    不覆盖 user/password/dbname，凭据仍只来自 env/*.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 关键字那种）：那种写法把分区列固定在
+    不再用 PARTITIONS 扩展语法（EXPLAIN PARTITIONS）：其分区列固定在结果 row[3]，列序随 MariaDB/MySQL
-    结果集第 4 列（下标 3），列序随 MariaDB/MySQL 版本变化，且该扩展语法在新版 MySQL 已废弃。
+    版本变化，且该扩展语法在新版 MySQL 已废弃。JSON 计划里的 partitions 字段两版一致。
-    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'"
+    planned = _explain_partitions(cur,
-                 " AND event_uid=%%s AND created_at=%%s" % TABLE)
+                                  "SELECT id FROM `%s` WHERE tenant_id='t_probe'"
-    planned = _explain_partitions(cur, probe_sql, (uid, ts))
+                                  " 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 已废弃该语法。
+    # 旧写法（PARTITIONS 扩展语法 + 取 row[3]）：列序跨版本不稳定，新版 MySQL 已废弃该语法。
-    prune_sql = ("SELECT id FROM `%s` WHERE created_at>='2026-01-01'"
+    pruned = _explain_partitions(cur,
-                 " AND created_at<'2026-02-01'" % TABLE)
+                                 "SELECT id FROM `%s` WHERE created_at>='2026-01-01'"
-    pruned = _explain_partitions(cur, prune_sql)
+                                 " 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)),