139 lines
4.4 KiB
Python
139 lines
4.4 KiB
Python
|
"""只读探查：`fin_risk_alert.trigger_rule_codes` 能否用 JSON 多值索引（docs/25 P3 #21）。
|
|||
|
|
|
|||
|
|
只做 SELECT / EXPLAIN，不做任何写入。结果写成 JSON 便于在 GBK 控制台下查看：
|
|||
|
|
|
|||
|
|
python tools/probe_risk_index.py
|
|||
|
|
|
|||
|
|
排查点：
|
|||
|
|
1. MySQL 版本是否 ≥ 8.0.17（多值索引的下限）；
|
|||
|
|
2. 列是否确实是 JSON、是否已存在同类索引；
|
|||
|
|
3. 是否存在**非数组**取值 —— 有的话 `CAST(... AS CHAR ARRAY)` 建索引会直接失败；
|
|||
|
|
4. 当前 `JSON_CONTAINS` 查询走的是什么访问路径（有没有可用的 key）。
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
import asyncio
|
|||
|
|
import json
|
|||
|
|
from pathlib import Path
|
|||
|
|
from typing import Any
|
|||
|
|
|
|||
|
|
from sqlalchemy import text
|
|||
|
|
|
|||
|
|
from app.infrastructure.db import SessionFactory
|
|||
|
|
|
|||
|
|
OUTPUT = Path("docs/evidence/risk-index-probe.json")
|
|||
|
|
|
|||
|
|
|
|||
|
|
async def collect() -> dict[str, Any]:
|
|||
|
|
report: dict[str, Any] = {}
|
|||
|
|
async with SessionFactory() as session:
|
|||
|
|
report["mysql_version"] = (
|
|||
|
|
await session.execute(text("SELECT VERSION()"))
|
|||
|
|
).scalar_one()
|
|||
|
|
|
|||
|
|
report["column"] = [
|
|||
|
|
dict(row)
|
|||
|
|
for row in (
|
|||
|
|
(
|
|||
|
|
await session.execute(
|
|||
|
|
text(
|
|||
|
|
"""
|
|||
|
|
SELECT COLUMN_NAME, COLUMN_TYPE, DATA_TYPE, IS_NULLABLE
|
|||
|
|
FROM information_schema.COLUMNS
|
|||
|
|
WHERE TABLE_SCHEMA = DATABASE()
|
|||
|
|
AND TABLE_NAME = 'fin_risk_alert'
|
|||
|
|
AND COLUMN_NAME = 'trigger_rule_codes'
|
|||
|
|
"""
|
|||
|
|
)
|
|||
|
|
)
|
|||
|
|
)
|
|||
|
|
.mappings()
|
|||
|
|
.all()
|
|||
|
|
)
|
|||
|
|
]
|
|||
|
|
|
|||
|
|
report["indexes"] = [
|
|||
|
|
dict(row)
|
|||
|
|
for row in (
|
|||
|
|
(
|
|||
|
|
await session.execute(
|
|||
|
|
text(
|
|||
|
|
"""
|
|||
|
|
SELECT INDEX_NAME, COLUMN_NAME, INDEX_TYPE, EXPRESSION
|
|||
|
|
FROM information_schema.STATISTICS
|
|||
|
|
WHERE TABLE_SCHEMA = DATABASE()
|
|||
|
|
AND TABLE_NAME = 'fin_risk_alert'
|
|||
|
|
ORDER BY INDEX_NAME, SEQ_IN_INDEX
|
|||
|
|
"""
|
|||
|
|
)
|
|||
|
|
)
|
|||
|
|
)
|
|||
|
|
.mappings()
|
|||
|
|
.all()
|
|||
|
|
)
|
|||
|
|
]
|
|||
|
|
|
|||
|
|
report["row_count"] = (
|
|||
|
|
await session.execute(text("SELECT COUNT(*) FROM fin_risk_alert"))
|
|||
|
|
).scalar_one()
|
|||
|
|
report["non_array_rows"] = (
|
|||
|
|
await session.execute(
|
|||
|
|
text(
|
|||
|
|
"SELECT COUNT(*) FROM fin_risk_alert "
|
|||
|
|
"WHERE trigger_rule_codes IS NULL "
|
|||
|
|
" OR JSON_TYPE(trigger_rule_codes) <> 'ARRAY'"
|
|||
|
|
)
|
|||
|
|
)
|
|||
|
|
).scalar_one()
|
|||
|
|
report["distinct_rule_codes"] = [
|
|||
|
|
dict(row)
|
|||
|
|
for row in (
|
|||
|
|
(
|
|||
|
|
await session.execute(
|
|||
|
|
text(
|
|||
|
|
"""
|
|||
|
|
SELECT CAST(trigger_rule_codes AS CHAR) AS raw_codes,
|
|||
|
|
COUNT(*) AS rows_count
|
|||
|
|
FROM fin_risk_alert
|
|||
|
|
GROUP BY 1
|
|||
|
|
"""
|
|||
|
|
)
|
|||
|
|
)
|
|||
|
|
)
|
|||
|
|
.mappings()
|
|||
|
|
.all()
|
|||
|
|
)
|
|||
|
|
]
|
|||
|
|
|
|||
|
|
# 用 FORMAT=JSON：传统 EXPLAIN 会把"该查询无法被缓存"写成 warning 打到 stderr，
|
|||
|
|
# 让脚本在 CI 里看起来是失败的，而它其实成功了。
|
|||
|
|
raw_explain = (
|
|||
|
|
await session.execute(
|
|||
|
|
text(
|
|||
|
|
"""
|
|||
|
|
EXPLAIN FORMAT=JSON
|
|||
|
|
SELECT id FROM fin_risk_alert
|
|||
|
|
WHERE JSON_CONTAINS(trigger_rule_codes, '"RW-018"')
|
|||
|
|
"""
|
|||
|
|
)
|
|||
|
|
)
|
|||
|
|
).scalar_one()
|
|||
|
|
report["explain_json_contains"] = json.loads(raw_explain)
|
|||
|
|
|
|||
|
|
return report
|
|||
|
|
|
|||
|
|
|
|||
|
|
async def main() -> None:
|
|||
|
|
report = await collect()
|
|||
|
|
OUTPUT.parent.mkdir(parents=True, exist_ok=True)
|
|||
|
|
OUTPUT.write_text(
|
|||
|
|
json.dumps(report, ensure_ascii=False, indent=2, default=str),
|
|||
|
|
encoding="utf-8",
|
|||
|
|
)
|
|||
|
|
print(f"wrote {OUTPUT}")
|
|||
|
|
|
|||
|
|
|
|||
|
|
if __name__ == "__main__":
|
|||
|
|
asyncio.run(main())
|