4. SQL变更的CI静态检查流水线搭建

为什么 SQL 变更需要独立的静态检查层

SQL 文件进入仓库后，如果只靠人工 Review，很容易漏掉方言不兼容、语法错误和破坏性操作。CI 静态检查流水线的价值在于：在合并请求（MR）阶段自动完成三类判断——语法能否解析、方言是否匹配目标引擎、是否包含 DROP/TRUNCATE 等危险语句，并把结果写成报告，不合规则直接阻断合并。

流水线核心实现

下面给出一个可直接运行的检查脚本，使用 sqlglot 做语法解析与方言校验，用正则匹配危险操作。脚本接收 SQL 文件路径列表，输出 JSON 报告，并以退出码表示是否阻断。

# ci_sql_check.py
import json
import re
import sys
from pathlib import Path

import sqlglot
from sqlglot.errors import ParseError

# 危险操作模式：DROP TABLE / TRUNCATE / DELETE 无 WHERE
DANGEROUS_PATTERNS = [
    (re.compile(r"\bDROP\s+TABLE\b", re.IGNORECASE), "DROP TABLE"),
    (re.compile(r"\bTRUNCATE\s+TABLE\b", re.IGNORECASE), "TRUNCATE TABLE"),
    (re.compile(r"\bDELETE\s+FROM\b(?!.*\bWHERE\b)", re.IGNORECASE | re.DOTALL), "DELETE 无 WHERE"),
]

# 目标方言映射
DIALECT_MAP = {
    "hive": "hive",
    "postgresql": "postgres",
    "mysql": "mysql",
}

def check_file(path: Path, dialect: str) -> dict:
    """对单个 SQL 文件执行三项检查，返回结果字典。"""
    result = {
        "file": str(path),
        "dialect": dialect,
        "syntax_ok": True,
        "syntax_error": None,
        "dialect_ok": True,
        "dialect_error": None,
        "dangerous": [],
    }
    sql_text = path.read_text(encoding="utf-8")

    # ① 语法解析
    try:
        sqlglot.parse(sql_text, read=dialect)
    except ParseError as e:
        result["syntax_ok"] = False
        result["syntax_error"] = str(e)
        return result

    # ② 方言校验：用目标方言重新解析，捕获不兼容
    try:
        sqlglot.parse(sql_text, read=dialect)
    except Exception as e:
        result["dialect_ok"] = False
        result["dialect_error"] = str(e)

    # ③ 危险操作拦截
    for pattern, label in DANGEROUS_PATTERNS:
        if pattern.search(sql_text):
            result["dangerous"].append(label)

    return result

def main():
    if len(sys.argv) < 3:
        print("用法: python ci_sql_check.py <dialect> <sql_file...>")
        sys.exit(2)

    dialect = DIALECT_MAP.get(sys.argv[1].lower())
    if not dialect:
        print(f"不支持的方言: {sys.argv[1]}，可选: {list(DIALECT_MAP)}")
        sys.exit(2)

    files = [Path(p) for p in sys.argv[2:]]
    reports = [check_file(f, dialect) for f in files]

    # 汇总
    errors = sum(1 for r in reports if not r["syntax_ok"] or not r["dialect_ok"])
    blocked = sum(1 for r in reports if r["dangerous"])

    report = {
        "dialect": dialect,
        "total": len(reports),
        "errors": errors,
        "blocked": blocked,
        "details": reports,
    }

    Path("sql_check_report.json").write_text(
        json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8"
    )

    # 控制台摘要
    for r in reports:
        status = "PASS"
        if not r["syntax_ok"]:
            status = "SYNTAX_ERROR"
        elif not r["dialect_ok"]:
            status = "DIALECT_ERROR"
        elif r["dangerous"]:
            status = "BLOCKED"
        print(f"[{status}] {r['file']}")
        if r["syntax_error"]:
            print(f"  语法: {r['syntax_error']}")
        if r["dialect_error"]:
            print(f"  方言: {r['dialect_error']}")
        if r["dangerous"]:
            print(f"  危险操作: {', '.join(r['dangerous'])}")

    # 退出码：有错误或阻断则非零
    sys.exit(1 if (errors or blocked) else 0)

if __name__ == "__main__":
    main()

CI 配置与门禁规则

在 GitLab CI 或 GitHub Actions 中，把上述脚本挂到 MR 的 test 阶段，并设置 allow_failure: false，即可实现阻断。以下为 GitLab CI 示例：

# .gitlab-ci.yml 片段
stages:
  - test

sql-static-check:
  stage: test
  image: python:3.11-slim
  before_script:
    - pip install sqlglot==25.0.0
  script:
    - |
      CHANGED=$(git diff --name-only origin/main...HEAD -- '*.sql')
      if [ -z "$CHANGED" ]; then
        echo "无 SQL 变更，跳过检查"
        exit 0
      fi
      python ci_sql_check.py hive $CHANGED
  artifacts:
    when: always
    paths:
      - sql_check_report.json
    expire_in: 7 days
  allow_failure: false

验证与结果判读

准备三个测试文件，覆盖通过、语法错误、危险操作三种情况：

-- test_pass.sql
SELECT id, name FROM users WHERE dt = '2024-01-01';

-- test_syntax_error.sql
SELEC id FROM users;

-- test_dangerous.sql
DROP TABLE users;

执行 python ci_sql_check.py hive test_pass.sql test_syntax_error.sql test_dangerous.sql，预期控制台输出：

[PASS] test_pass.sql
[SYNTAX_ERROR] test_syntax_error.sql
  语法: Invalid expression / Unexpected token. Line 1, Col: 1.
[BLOCKED] test_dangerous.sql
  危险操作: DROP TABLE

退出码为 1，CI 任务失败，MR 被阻断。打开 sql_check_report.json 可看到每条 SQL 的详细结果，评论机器人可将 details 中的 file、syntax_error、dangerous 回写到 MR 评论，方便开发者定位。

可直接使用的检查清单