新增 SQL 题型：SQLite 内联判题（不依赖外部沙箱）

- SQL 作为语言接入现有提交流程，judge_task 按 language 分流到 SQLJudgeDispatcher
- judge/sql_runner.py：内存 SQLite 判题核心，查询题比结果集/增删改题比表状态，
  authorizer + progress_handler + max_page_count 三重防护
- dispatcher 提取 _process_judge_result/_push_status 供 SQL 判题复用（行为不变）
- 测试点通道支持 1.sql..N.sql 压缩包，出题只需数据脚本+标准答案
- Problem 增加 sql_config 字段；options 数据迁移注册 SQL 语言

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

judge/dispatcher.py

@@ -28,6 +28,7 @@ def process_pending_task():
     if cache.llen(CacheKey.waiting_queue):
         # 防止循环引入
         from judge.tasks import judge_task
+
         tmp_data = cache.rpop(CacheKey.waiting_queue)
         if tmp_data:
             data = json.loads(tmp_data.decode("utf-8"))
@@ -42,8 +43,7 @@ class ChooseJudgeServer:
         with transaction.atomic():
             cutoff = timezone.now() - timedelta(seconds=6)
             server = (
-                JudgeServer.objects
-                .select_for_update(skip_locked=True)
+                JudgeServer.objects.select_for_update(skip_locked=True)
                 .filter(
                     is_disabled=False,
                     last_heartbeat__gte=cutoff,
@@ -78,8 +78,6 @@ class DispatcherBase(object):
             logger.exception(e)
 
 
-
-
 class JudgeDispatcher(DispatcherBase):
     def __init__(self, submission_id, problem_id):
         super().__init__()
@@ -93,6 +91,24 @@ class JudgeDispatcher(DispatcherBase):
         else:
             self.problem = Problem.objects.get(id=problem_id)
 
+    def _push_status(self, result, status, extra=None):
+        data = {
+            "type": "submission_update",
+            "submission_id": str(self.submission.id),
+            "result": result,
+            "status": status,
+        }
+        if extra:
+            data.update(extra)
+        try:
+            push_submission_update(
+                submission_id=str(self.submission.id),
+                user_id=self.submission.user_id,
+                data=data,
+            )
+        except Exception as e:
+            logger.error(f"Failed to push submission update: {str(e)}")
+
     def _compute_statistic_info(self, resp_data):
         # 用时和内存占用保存为多个测试点中最长的那个
         self.submission.statistic_info["time_cost"] = max([x["cpu_time"] for x in resp_data])
@@ -131,7 +147,7 @@ class JudgeDispatcher(DispatcherBase):
             "max_memory": 1024 * 1024 * self.problem.memory_limit,
             "test_case_id": self.problem.test_case_id,
             "output": False,
-            "io_mode": self.problem.io_mode
+            "io_mode": self.problem.io_mode,
         }
 
         with ChooseJudgeServer() as server:
@@ -139,57 +155,30 @@ class JudgeDispatcher(DispatcherBase):
                 data = {"submission_id": self.submission.id, "problem_id": self.problem.id}
                 cache.lpush(CacheKey.waiting_queue, json.dumps(data))
                 # 推送排队状态
-                try:
-                    push_submission_update(
-                        submission_id=str(self.submission.id),
-                        user_id=self.submission.user_id,
-                        data={
-                            "type": "submission_update",
-                            "submission_id": str(self.submission.id),
-                            "result": JudgeStatus.PENDING,
-                            "status": "pending",
-                        }
-                    )
-                except Exception as e:
-                    logger.error(f"Failed to push submission update: {str(e)}")
+                self._push_status(JudgeStatus.PENDING, "pending")
                 return
             Submission.objects.filter(id=self.submission.id).update(result=JudgeStatus.JUDGING)
 
             # 推送判题中状态
-            try:
-                push_submission_update(
-                    submission_id=str(self.submission.id),
-                    user_id=self.submission.user_id,
-                    data={
-                        "type": "submission_update",
-                        "submission_id": str(self.submission.id),
-                        "result": JudgeStatus.JUDGING,
-                        "status": "judging",
-                    }
-                )
-            except Exception as e:
-                logger.error(f"Failed to push submission update: {str(e)}")
+            self._push_status(JudgeStatus.JUDGING, "judging")
 
             resp = self._request(urljoin(server.service_url, "/judge"), data=data)
 
         if not resp:
             Submission.objects.filter(id=self.submission.id).update(result=JudgeStatus.SYSTEM_ERROR)
             # 推送系统错误状态
-            try:
-                push_submission_update(
-                    submission_id=str(self.submission.id),
-                    user_id=self.submission.user_id,
-                    data={
-                        "type": "submission_update",
-                        "submission_id": str(self.submission.id),
-                        "result": JudgeStatus.SYSTEM_ERROR,
-                        "status": "error",
-                    }
-                )
-            except Exception as e:
-                logger.error(f"Failed to push submission update: {str(e)}")
+            self._push_status(JudgeStatus.SYSTEM_ERROR, "error")
             return
 
+        self._process_judge_result(resp)
+
+    def _process_judge_result(self, resp):
+        """判题结果的统一后处理：状态聚合、AST 钩子、统计、排名、WebSocket 推送。
+
+        resp 结构与外部 judger 返回一致：{"err": "CompileError"|None, "data": ...}；
+        SQLJudgeDispatcher 组装同构 resp 后也走这里。
+        """
+        language = self.submission.language
         if resp["err"]:
             self.submission.result = JudgeStatus.COMPILE_ERROR
             self.submission.statistic_info["err_info"] = resp["data"]
@@ -220,31 +209,22 @@ class JudgeDispatcher(DispatcherBase):
         self.submission.save(update_fields=["result", "info", "statistic_info"])
 
         # 推送判题完成状态
-        try:
-            push_submission_update(
-                submission_id=str(self.submission.id),
-                user_id=self.submission.user_id,
-                data={
-                    "type": "submission_update",
-                    "submission_id": str(self.submission.id),
-                    "result": self.submission.result,
-                    "status": "finished",
-                    "time_cost": self.submission.statistic_info.get("time_cost"),
-                    "memory_cost": self.submission.statistic_info.get("memory_cost"),
-                    "score": self.submission.statistic_info.get("score", 0),
-                }
-            )
-        except Exception as e:
-            logger.error(f"Failed to push submission update: {str(e)}")
+        self._push_status(
+            self.submission.result,
+            "finished",
+            extra={
+                "time_cost": self.submission.statistic_info.get("time_cost"),
+                "memory_cost": self.submission.statistic_info.get("memory_cost"),
+                "score": self.submission.statistic_info.get("score", 0),
+            },
+        )
 
         if self.contest_id:
             # 以提交时刻（而非判题时刻）是否落在比赛时间窗内为准，
             # 避免临界提交因判题排队延迟到比赛结束后才处理而被丢弃
             in_contest = self.contest.start_time <= self.submission.create_time <= self.contest.end_time
-            if not in_contest or \
-                    User.objects.get(id=self.submission.user_id).is_contest_admin(self.contest):
-                logger.info(
-                    "Contest debug mode, id: " + str(self.contest_id) + ", submission id: " + self.submission.id)
+            if not in_contest or User.objects.get(id=self.submission.user_id).is_contest_admin(self.contest):
+                logger.info("Contest debug mode, id: " + str(self.contest_id) + ", submission id: " + self.submission.id)
                 return
             with transaction.atomic():
                 self.update_contest_problem_status()
@@ -286,8 +266,7 @@ class JudgeDispatcher(DispatcherBase):
                 score = self.submission.statistic_info["score"]
                 if not is_accepted(oi_problems_status[problem_id]["status"]):
                     # minus last time score, add this tim score
-                    profile.add_score(this_time_score=score,
-                                      last_time_score=oi_problems_status[problem_id]["score"])
+                    profile.add_score(this_time_score=score, last_time_score=oi_problems_status[problem_id]["score"])
                     oi_problems_status[problem_id]["score"] = score
                     oi_problems_status[problem_id]["status"] = JudgeStatus.ACCEPTED if is_accepted(self.submission.result) else self.submission.result
                     if is_accepted(self.submission.result):
@@ -331,15 +310,12 @@ class JudgeDispatcher(DispatcherBase):
                 score = self.submission.statistic_info["score"]
                 if problem_id not in oi_problems_status:
                     user_profile.add_score(score)
-                    oi_problems_status[problem_id] = {"status": profile_status,
-                                                      "_id": self.problem._id,
-                                                      "score": score}
+                    oi_problems_status[problem_id] = {"status": profile_status, "_id": self.problem._id, "score": score}
                     if is_accepted(self.submission.result):
                         user_profile.accepted_number += 1
                 elif not is_accepted(oi_problems_status[problem_id]["status"]):
                     # minus last time score, add this time score
-                    user_profile.add_score(this_time_score=score,
-                                           last_time_score=oi_problems_status[problem_id]["score"])
+                    user_profile.add_score(this_time_score=score, last_time_score=oi_problems_status[problem_id]["score"])
                     oi_problems_status[problem_id]["score"] = score
                     oi_problems_status[problem_id]["status"] = profile_status
                     if is_accepted(self.submission.result):
@@ -425,4 +401,3 @@ class JudgeDispatcher(DispatcherBase):
                 info["error_number"] = 1
         rank.submission_info[str(self.submission.problem_id)] = info
         rank.save(update_fields=["submission_info", "total_time", "accepted_number", "submission_number"])
-

judge/languages.py

@@ -174,4 +174,6 @@ languages = [
     {"config": _py3_lang_config, "name": "Python3", "description": "Python 3.12", "content_type": "text/x-python"},
     {"config": _go_lang_config, "name": "Golang", "description": "Golang 1.22", "content_type": "text/x-go"},
     {"config": _node_lang_config, "name": "JavaScript", "description": "Node.js 20", "content_type": "text/javascript"},
+    # SQL 题不走外部 judger 沙箱（judge/sql_dispatcher.py 在 worker 内用 sqlite3 判题），config 仅占位
+    {"config": {"template": ""}, "name": "SQL", "description": "SQLite 3", "content_type": "text/x-sql"},
 ]

judge/sql_dispatcher.py

@@ -0,0 +1,101 @@
+"""SQL 题判题调度：不走 JudgeServer 沙箱，在 dramatiq worker 内用 sqlite3 直接判题。
+
+复用 JudgeDispatcher 的 _process_judge_result 完成状态聚合、统计、排名和 WebSocket 推送，
+因此比赛排名、rejudge、题目统计的语义与沙箱判题完全一致。
+"""
+
+import json
+import logging
+import os
+
+from django.conf import settings
+
+from judge.dispatcher import JudgeDispatcher
+from judge.sql_runner import SQLCaseError, run_case
+from submission.models import JudgeStatus, Submission
+from utils.shortcuts import natural_sort_key
+
+logger = logging.getLogger(__name__)
+
+
+class SQLProblemConfigError(Exception):
+    pass
+
+
+class SQLJudgeDispatcher(JudgeDispatcher):
+    def judge(self):
+        Submission.objects.filter(id=self.submission.id).update(result=JudgeStatus.JUDGING)
+        self._push_status(JudgeStatus.JUDGING, "judging")
+
+        try:
+            ref_sql, mode, order_sensitive, init_scripts = self._load_problem_config()
+        except SQLProblemConfigError as e:
+            self._system_error(str(e))
+            return
+
+        cases = []
+        for index, init_sql in enumerate(init_scripts, start=1):
+            try:
+                case = run_case(
+                    init_sql,
+                    ref_sql,
+                    self.submission.code,
+                    mode=mode,
+                    order_sensitive=order_sensitive,
+                    time_limit_ms=self.problem.time_limit,
+                    memory_limit_mb=self.problem.memory_limit,
+                )
+            except SQLCaseError as e:
+                # 初始化/标准答案执行失败，属出题配置问题
+                self._system_error(e.message)
+                return
+            case["test_case"] = str(index)
+            # 语法错误与数据无关，首个测试点即可确认，整题按编译错误处理（ACM 不罚时，前端展示 err_info）
+            if index == 1 and case["result"] == JudgeStatus.COMPILE_ERROR:
+                self._process_judge_result({"err": "CompileError", "data": case["error_message"]})
+                return
+            cases.append(case)
+
+        self._process_judge_result({"err": None, "data": cases})
+
+    def _system_error(self, message):
+        logger.error(f"SQL judge system error, submission {self.submission.id}, problem {self.problem.id}: {message}")
+        Submission.objects.filter(id=self.submission.id).update(result=JudgeStatus.SYSTEM_ERROR, statistic_info={"err_info": message})
+        self._push_status(JudgeStatus.SYSTEM_ERROR, "error")
+
+    def _load_problem_config(self):
+        """校验并加载 SQL 题配置，返回 (标准答案, mode, order_sensitive, 各测试点初始化脚本)。"""
+        sql_config = self.problem.sql_config or {}
+        mode = sql_config.get("mode")
+        if mode not in ("query", "modify"):
+            raise SQLProblemConfigError("题目缺少 SQL 配置（题型）")
+
+        ref_sql = None
+        for item in self.problem.answers or []:
+            if item.get("language") == "SQL" and item.get("code", "").strip():
+                ref_sql = item["code"]
+                break
+        if not ref_sql:
+            raise SQLProblemConfigError("题目缺少 SQL 标准答案")
+
+        test_case_dir = os.path.join(settings.TEST_CASE_DIR, self.problem.test_case_id)
+        try:
+            with open(os.path.join(test_case_dir, "info"), encoding="utf-8") as f:
+                info = json.load(f)
+        except (OSError, json.JSONDecodeError) as e:
+            raise SQLProblemConfigError(f"测试点信息读取失败: {e}")
+        if not info.get("sql"):
+            raise SQLProblemConfigError("测试点不是 SQL 类型，请重新上传 SQL 测试点压缩包")
+
+        init_scripts = []
+        # 按 "1","2",… 自然序遍历，保证与 test_case_score 的下标对应（OI 计分依赖顺序）
+        for key in sorted(info["test_cases"].keys(), key=natural_sort_key):
+            input_name = info["test_cases"][key]["input_name"]
+            try:
+                with open(os.path.join(test_case_dir, input_name), encoding="utf-8") as f:
+                    init_scripts.append(f.read())
+            except OSError as e:
+                raise SQLProblemConfigError(f"测试点脚本 {input_name} 读取失败: {e}")
+        if not init_scripts:
+            raise SQLProblemConfigError("题目没有任何测试点")
+        return ref_sql, mode, sql_config.get("order_sensitive", False), init_scripts

judge/sql_runner.py

@@ -0,0 +1,246 @@
+"""SQL 题判题核心：在内存 SQLite 中分别执行标准答案和学生 SQL 并比对结果。
+
+查询题（mode="query"）比对最后一条 SELECT 的结果集；
+增删改题（mode="modify"）比对执行后所有用户表的最终状态。
+学生 SQL 通过 authorizer（禁 ATTACH/PRAGMA，查询题只读）、
+progress_handler（墙钟超时）和 max_page_count（内存上限）三重防护。
+"""
+
+import sqlite3
+import time
+
+from submission.models import JudgeStatus
+
+# 单结果集/单表最大行数，防 CROSS JOIN 撑爆 worker 内存
+ROW_LIMIT = 10_000
+# progress_handler 检查粒度（SQLite VM 指令数）
+PROGRESS_STEP = 1_000
+ERROR_MESSAGE_MAX_LEN = 200
+
+# prepare 阶段的语法类错误，映射为 COMPILE_ERROR
+_SYNTAX_ERROR_MARKERS = ("syntax error", "unrecognized token", "incomplete input")
+
+# 两种模式都禁止的授权码：挂载外部库 / 数据库参数
+_DENIED_ALWAYS = {sqlite3.SQLITE_ATTACH, sqlite3.SQLITE_DETACH, sqlite3.SQLITE_PRAGMA}
+
+# 查询题允许的授权码（白名单外一律拒绝，防先 INSERT 伪造数据再 SELECT）
+_QUERY_MODE_ALLOWED = {getattr(sqlite3, name) for name in ("SQLITE_SELECT", "SQLITE_READ", "SQLITE_FUNCTION", "SQLITE_RECURSIVE", "SQLITE_TRANSACTION") if hasattr(sqlite3, name)}
+
+
+class SQLCaseError(Exception):
+    """携带 JudgeStatus 的判题异常。SYSTEM_ERROR 级别（出题配置问题）会传播到 dispatcher。"""
+
+    def __init__(self, result, message):
+        super().__init__(message)
+        self.result = result
+        self.message = message
+
+
+def _truncate(message):
+    message = str(message)
+    if len(message) > ERROR_MESSAGE_MAX_LEN:
+        return message[:ERROR_MESSAGE_MAX_LEN] + "..."
+    return message
+
+
+def split_statements(script):
+    """用 sqlite3.complete_statement 累积切分多条语句，正确处理字符串/注释里的分号；末尾缺分号自动补。"""
+    statements = []
+    buf = ""
+    for part in script.split(";"):
+        buf += part + ";"
+        if sqlite3.complete_statement(buf):
+            stmt = buf.strip()
+            buf = ""
+            if stmt and stmt != ";":
+                statements.append(stmt)
+    # 残句（未闭合的引号/注释，或末尾缺分号但上面已补），交给 execute 报错或执行
+    remainder = buf.strip()
+    if remainder and remainder != ";":
+        statements.append(remainder)
+    return statements
+
+
+def _canonical_value(v):
+    """值归一化并打类型标签，防止 NULL/"NULL"、1/"1" 碰撞；数值统一比对（1 == 1.0，浮点 6 位有效数字）。"""
+    if v is None:
+        return ("null",)
+    if isinstance(v, int):
+        return ("num", str(v))
+    if isinstance(v, float):
+        if v.is_integer() and abs(v) < 2**53:
+            return ("num", str(int(v)))
+        return ("num", format(v, ".6g"))
+    if isinstance(v, bytes):
+        return ("blob", v.hex())
+    return ("str", str(v))
+
+
+def _canonical_row(row):
+    return tuple(_canonical_value(v) for v in row)
+
+
+def _new_db(memory_limit_mb):
+    conn = sqlite3.connect(":memory:", isolation_level=None)  # autocommit，脚本行为可预期
+    conn.execute("PRAGMA page_size=4096")
+    # 4096B/页 × 256 页/MB，超限报 "database or disk is full"
+    conn.execute(f"PRAGMA max_page_count={max(int(memory_limit_mb), 1) * 256}")
+    return conn
+
+
+def _execute_statements(conn, script, deadline=None):
+    """逐条执行，返回最后一条产生结果集的语句的 (列数, 行列表)；无结果集返回 None。"""
+    last_result = None
+    for stmt in split_statements(script):
+        if deadline is not None and time.monotonic() > deadline:
+            raise sqlite3.OperationalError("interrupted")
+        cursor = conn.execute(stmt)
+        if cursor.description is not None:
+            rows = cursor.fetchmany(ROW_LIMIT + 1)
+            if len(rows) > ROW_LIMIT:
+                raise SQLCaseError(JudgeStatus.MEMORY_LIMIT_EXCEEDED, f"查询结果超过 {ROW_LIMIT} 行")
+            last_result = (len(cursor.description), [_canonical_row(r) for r in rows])
+        cursor.close()
+    return last_result
+
+
+def _dump_tables(conn):
+    """dump 所有用户表：{表名: (列数, 行多重集)}，行内排序，表状态天然无序。"""
+    cursor = conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name")
+    tables = [r[0] for r in cursor.fetchall()]
+    state = {}
+    for table in tables:
+        quoted = table.replace('"', '""')
+        cur = conn.execute(f'SELECT * FROM "{quoted}"')
+        rows = cur.fetchmany(ROW_LIMIT + 1)
+        if len(rows) > ROW_LIMIT:
+            raise SQLCaseError(JudgeStatus.MEMORY_LIMIT_EXCEEDED, f"表 {table} 超过 {ROW_LIMIT} 行")
+        state[table] = (len(cur.description), sorted(_canonical_row(r) for r in rows))
+    return state
+
+
+def _execute_trusted(conn, script, deadline, error_prefix):
+    """执行受信脚本，任何失败都是出题问题 → SYSTEM_ERROR。"""
+    conn.set_progress_handler(lambda: 1 if time.monotonic() > deadline else 0, PROGRESS_STEP)
+    try:
+        return _execute_statements(conn, script, deadline)
+    except SQLCaseError as e:
+        raise SQLCaseError(JudgeStatus.SYSTEM_ERROR, f"{error_prefix}: {e.message}")
+    except sqlite3.Error as e:
+        raise SQLCaseError(JudgeStatus.SYSTEM_ERROR, f"{error_prefix}: {_truncate(e)}")
+    finally:
+        conn.set_progress_handler(None, PROGRESS_STEP)
+
+
+def _run_student(conn, script, mode, deadline):
+    """带三重防护执行学生 SQL，异常映射为学生级 JudgeStatus。"""
+    denied_hints = []
+
+    def authorizer(action, arg1, arg2, db_name, trigger):
+        if action in _DENIED_ALWAYS:
+            denied_hints.append("禁止使用 ATTACH/DETACH/PRAGMA 等数据库管理语句")
+            return sqlite3.SQLITE_DENY
+        if mode == "query" and action not in _QUERY_MODE_ALLOWED:
+            denied_hints.append("本题为查询题，禁止修改数据或表结构（INSERT/UPDATE/DELETE/CREATE 等）")
+            return sqlite3.SQLITE_DENY
+        return sqlite3.SQLITE_OK
+
+    conn.set_authorizer(authorizer)
+    conn.set_progress_handler(lambda: 1 if time.monotonic() > deadline else 0, PROGRESS_STEP)
+    try:
+        last_result = _execute_statements(conn, script, deadline)
+    except sqlite3.Error as e:
+        # 注意：authorizer 拒绝抛的是 DatabaseError 基类而非 OperationalError，统一按消息映射
+        msg = str(e)
+        if "interrupted" in msg:
+            raise SQLCaseError(JudgeStatus.CPU_TIME_LIMIT_EXCEEDED, "SQL 执行超时")
+        if "database or disk is full" in msg:
+            raise SQLCaseError(JudgeStatus.MEMORY_LIMIT_EXCEEDED, "数据量超出内存限制")
+        if "not authorized" in msg or "prohibited" in msg:
+            hint = denied_hints[-1] if denied_hints else "本题禁止使用该语句"
+            raise SQLCaseError(JudgeStatus.RUNTIME_ERROR, hint)
+        if any(marker in msg for marker in _SYNTAX_ERROR_MARKERS):
+            raise SQLCaseError(JudgeStatus.COMPILE_ERROR, _truncate(msg))
+        raise SQLCaseError(JudgeStatus.RUNTIME_ERROR, _truncate(msg))
+    finally:
+        conn.set_progress_handler(None, PROGRESS_STEP)
+        conn.set_authorizer(None)
+
+    if mode == "query":
+        return last_result
+    # dump 是我们自己的读取，不应吃学生的超时/授权限制（上面已清除）
+    return _dump_tables(conn)
+
+
+def _compare(expected, actual, mode, order_sensitive):
+    if mode == "query":
+        exp_cols, exp_rows = expected
+        act_cols, act_rows = actual
+        if exp_cols != act_cols:
+            return False
+        if order_sensitive:
+            return exp_rows == act_rows
+        return sorted(exp_rows) == sorted(act_rows)
+    # modify: dump dict 里行已排序
+    return expected == actual
+
+
+def run_case(init_sql, ref_sql, student_sql, *, mode, order_sensitive, time_limit_ms, memory_limit_mb):
+    """判一个测试点，返回与外部 judger 单测试点同构的 dict。
+
+    学生错误（CE/WA/TLE/MLE/RE）体现在返回值里；
+    出题配置错误（初始化/标准答案失败）抛 SQLCaseError(SYSTEM_ERROR)，由 dispatcher 处理。
+    """
+    time_limit_s = time_limit_ms / 1000
+    # 受信脚本（初始化/标准答案）的运行上限放宽，避免出题数据较大时误报；仍防 worker 永久阻塞
+    trusted_limit_s = max(time_limit_s * 5, 10)
+
+    ref_conn = _new_db(memory_limit_mb)
+    try:
+        _execute_trusted(ref_conn, init_sql, time.monotonic() + trusted_limit_s, "初始化脚本执行失败")
+        last_result = _execute_trusted(ref_conn, ref_sql, time.monotonic() + trusted_limit_s, "标准答案执行失败")
+        if mode == "query":
+            expected = last_result
+        else:
+            try:
+                expected = _dump_tables(ref_conn)
+            except SQLCaseError as e:
+                raise SQLCaseError(JudgeStatus.SYSTEM_ERROR, f"标准答案结果超出限制: {e.message}")
+    finally:
+        ref_conn.close()
+    if mode == "query" and expected is None:
+        raise SQLCaseError(JudgeStatus.SYSTEM_ERROR, "标准答案未产生查询结果集")
+
+    case = {
+        "test_case": "",
+        "result": JudgeStatus.ACCEPTED,
+        "cpu_time": 0,
+        "real_time": 0,
+        "memory": 0,
+        "signal": 0,
+        "exit_code": 0,
+        "error": 0,
+        "output_md5": "",
+        "error_message": None,
+    }
+
+    stu_conn = _new_db(memory_limit_mb)
+    try:
+        _execute_trusted(stu_conn, init_sql, time.monotonic() + trusted_limit_s, "初始化脚本执行失败")
+        start = time.monotonic()
+        try:
+            actual = _run_student(stu_conn, student_sql, mode, start + time_limit_s)
+        except SQLCaseError as e:
+            elapsed = int((time.monotonic() - start) * 1000)
+            case.update(result=e.result, error_message=e.message, cpu_time=elapsed, real_time=elapsed)
+            return case
+        elapsed = int((time.monotonic() - start) * 1000)
+    finally:
+        stu_conn.close()
+
+    case["cpu_time"] = case["real_time"] = elapsed
+    if mode == "query" and actual is None:
+        case.update(result=JudgeStatus.WRONG_ANSWER, error_message="提交的 SQL 未产生查询结果集")
+    elif not _compare(expected, actual, mode, order_sensitive):
+        case["result"] = JudgeStatus.WRONG_ANSWER
+    return case

judge/tasks.py

@@ -2,13 +2,18 @@ import dramatiq
 
 from account.models import User
 from judge.dispatcher import JudgeDispatcher
+from judge.sql_dispatcher import SQLJudgeDispatcher
 from submission.models import Submission
 from utils.shortcuts import DRAMATIQ_WORKER_ARGS
 
 
 @dramatiq.actor(**DRAMATIQ_WORKER_ARGS())
 def judge_task(submission_id, problem_id):
-    uid = Submission.objects.get(id=submission_id).user_id
-    if User.objects.get(id=uid).is_disabled:
+    submission = Submission.objects.get(id=submission_id)
+    if User.objects.get(id=submission.user_id).is_disabled:
         return
-    JudgeDispatcher(submission_id, problem_id).judge()
+    # SQL 题不依赖 JudgeServer 沙箱，在 worker 内用 sqlite3 判题
+    if submission.language == "SQL":
+        SQLJudgeDispatcher(submission_id, problem_id).judge()
+    else:
+        JudgeDispatcher(submission_id, problem_id).judge()

options/migrations/0002_add_sql_language.py

@@ -0,0 +1,43 @@
+# SysOptions.languages 存在数据库里，_init_option 只在 key 缺失时写入默认值，
+# 因此新增 SQL 语言必须用数据迁移追加到已部署库；不用 reset_languages()（会覆盖管理员的自定义配置）。
+
+from django.db import migrations
+
+SQL_LANGUAGE = {
+    "config": {"template": ""},
+    "name": "SQL",
+    "description": "SQLite 3",
+    "content_type": "text/x-sql",
+}
+
+
+def add_sql_language(apps, schema_editor):
+    SysOptions = apps.get_model("options", "SysOptions")
+    try:
+        option = SysOptions.objects.get(key="languages")
+    except SysOptions.DoesNotExist:
+        # 库还没初始化过 languages，留给 _init_option 用代码默认值（已含 SQL）创建
+        return
+    if not any(item.get("name") == "SQL" for item in option.value):
+        option.value.append(SQL_LANGUAGE)
+        option.save(update_fields=["value"])
+
+
+def remove_sql_language(apps, schema_editor):
+    SysOptions = apps.get_model("options", "SysOptions")
+    try:
+        option = SysOptions.objects.get(key="languages")
+    except SysOptions.DoesNotExist:
+        return
+    option.value = [item for item in option.value if item.get("name") != "SQL"]
+    option.save(update_fields=["value"])
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("options", "0001_initial"),
+    ]
+
+    operations = [
+        migrations.RunPython(add_sql_language, remove_sql_language),
+    ]

problem/migrations/0011_problem_sql_config.py

@@ -0,0 +1,18 @@
+# Generated by Django 6.0.4 on 2026-07-02 15:57
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('problem', '0010_problem_ast_rules'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='problem',
+            name='sql_config',
+            field=models.JSONField(blank=True, default=None, null=True),
+        ),
+    ]

problem/models.py

@@ -86,6 +86,9 @@ class Problem(models.Model):
     # AST 代码结构检查规则
     ast_rules = models.JSONField(null=True, blank=True, default=None)
 
+    # SQL 题配置: {"mode": "query"|"modify", "order_sensitive": bool}，非 SQL 题为 None
+    sql_config = models.JSONField(null=True, blank=True, default=None)
+
     class Meta:
         db_table = "problem"
         constraints = [

problem/serializers.py

@@ -37,6 +37,11 @@ class CreateProblemCodeTemplateSerializer(serializers.Serializer):
     pass
 
 
+class SQLConfigSerializer(serializers.Serializer):
+    mode = serializers.ChoiceField(choices=["query", "modify"])
+    order_sensitive = serializers.BooleanField(default=False)
+
+
 class ProblemIOModeSerializer(serializers.Serializer):
     io_mode = serializers.ChoiceField(choices=ProblemIOMode.choices)
     input = serializers.CharField()
@@ -89,6 +94,9 @@ class CreateOrEditProblemSerializer(serializers.Serializer):
     # AST 规则
     ast_rules = serializers.JSONField(required=False, allow_null=True, default=None)
 
+    # SQL 题配置
+    sql_config = SQLConfigSerializer(required=False, allow_null=True, default=None)
+
 
 class CreateProblemSerializer(CreateOrEditProblemSerializer):
     pass

problem/views/admin.py

@@ -33,13 +33,16 @@ from ..serializers import (
 
 
 class TestCaseZipProcessor(object):
-    def process_zip(self, uploaded_zip_file, dir=""):
+    def process_zip(self, uploaded_zip_file, dir="", sql=False):
         try:
             zip_file = zipfile.ZipFile(uploaded_zip_file, "r")
         except zipfile.BadZipFile:
             raise APIError("Bad zip file")
         name_list = zip_file.namelist()
-        test_case_list = self.filter_name_list(name_list, dir=dir)
+        if sql:
+            test_case_list = self.filter_sql_name_list(name_list, dir=dir)
+        else:
+            test_case_list = self.filter_name_list(name_list, dir=dir)
         if not test_case_list:
             raise APIError("Empty file")
 
@@ -62,18 +65,33 @@ class TestCaseZipProcessor(object):
 
         info = []
 
-        # ["1.in", "1.out", "2.in", "2.out"] => [("1.in", "1.out"), ("2.in", "2.out")]
-        test_case_list = zip(*[test_case_list[i::2] for i in range(2)])
-        for index, item in enumerate(test_case_list):
-            data = {
-                "stripped_output_md5": md5_cache[item[1]],
-                "input_size": size_cache[item[0]],
-                "output_size": size_cache[item[1]],
-                "input_name": item[0],
-                "output_name": item[1],
-            }
-            info.append(data)
-            test_case_info["test_cases"][str(index + 1)] = data
+        if sql:
+            # SQL 题：每个 N.sql 是一个测试点的建表+数据脚本，没有期望输出（判题时跑标准答案生成）。
+            # output_name 复用同名、md5 置空，以兼容 CreateTestCaseScoreSerializer 和前端测试点表格。
+            test_case_info["sql"] = True
+            for index, item in enumerate(test_case_list):
+                data = {
+                    "stripped_output_md5": "",
+                    "input_size": size_cache[item],
+                    "output_size": 0,
+                    "input_name": item,
+                    "output_name": item,
+                }
+                info.append(data)
+                test_case_info["test_cases"][str(index + 1)] = data
+        else:
+            # ["1.in", "1.out", "2.in", "2.out"] => [("1.in", "1.out"), ("2.in", "2.out")]
+            test_case_list = zip(*[test_case_list[i::2] for i in range(2)])
+            for index, item in enumerate(test_case_list):
+                data = {
+                    "stripped_output_md5": md5_cache[item[1]],
+                    "input_size": size_cache[item[0]],
+                    "output_size": size_cache[item[1]],
+                    "input_name": item[0],
+                    "output_name": item[1],
+                }
+                info.append(data)
+                test_case_info["test_cases"][str(index + 1)] = data
 
         with open(os.path.join(test_case_dir, "info"), "w", encoding="utf-8") as f:
             f.write(json.dumps(test_case_info, indent=4))
@@ -97,6 +115,19 @@ class TestCaseZipProcessor(object):
             else:
                 return sorted(ret, key=natural_sort_key)
 
+    def filter_sql_name_list(self, name_list, dir=""):
+        # SQL 题测试点：连续编号的 1.sql, 2.sql, ...
+        ret = []
+        prefix = 1
+        while True:
+            name = f"{prefix}.sql"
+            if f"{dir}{name}" in name_list:
+                ret.append(name)
+                prefix += 1
+                continue
+            else:
+                return sorted(ret, key=natural_sort_key)
+
 
 class TestCaseAPI(CSRFExemptAPIView, TestCaseZipProcessor):
     request_parsers = ()
@@ -118,7 +149,19 @@ class TestCaseAPI(CSRFExemptAPIView, TestCaseZipProcessor):
         test_case_dir = os.path.join(settings.TEST_CASE_DIR, problem.test_case_id)
         if not os.path.isdir(test_case_dir):
             return self.error("Test case does not exists")
-        name_list = self.filter_name_list(os.listdir(test_case_dir))
+        # SQL 题的测试点是 N.sql，需按 info 里的类型标记选择文件列表
+        is_sql = False
+        info_path = os.path.join(test_case_dir, "info")
+        if os.path.isfile(info_path):
+            try:
+                with open(info_path, encoding="utf-8") as f:
+                    is_sql = bool(json.load(f).get("sql"))
+            except (OSError, json.JSONDecodeError):
+                pass
+        if is_sql:
+            name_list = self.filter_sql_name_list(os.listdir(test_case_dir))
+        else:
+            name_list = self.filter_name_list(os.listdir(test_case_dir))
         name_list.append("info")
         file_name = os.path.join(test_case_dir, problem.test_case_id + ".zip")
         with zipfile.ZipFile(file_name, "w") as file:
@@ -140,7 +183,8 @@ class TestCaseAPI(CSRFExemptAPIView, TestCaseZipProcessor):
         with open(zip_file, "wb") as f:
             for chunk in file:
                 f.write(chunk)
-        info, test_case_id = self.process_zip(zip_file)
+        sql = request.POST.get("sql") in ("1", "true", "True")
+        info, test_case_id = self.process_zip(zip_file, sql=sql)
         os.remove(zip_file)
         return self.success({"id": test_case_id, "info": info})
 
@@ -158,6 +202,19 @@ class ProblemBase(APIView):
             data["total_score"] = total_score
         data["languages"] = list(data["languages"])
 
+        # SQL 题校验：.sql 测试点与 .in/.out 沙箱判题互斥，SQL 必须是唯一语言
+        if "SQL" in data["languages"]:
+            if data["languages"] != ["SQL"]:
+                return "SQL problem cannot be mixed with other languages"
+            if not data.get("sql_config"):
+                return "SQL problem requires sql_config"
+            has_sql_answer = any(item.get("language") == "SQL" and item.get("code", "").strip() for item in (data.get("answers") or []))
+            if not has_sql_answer:
+                return "SQL problem requires a SQL reference answer"
+        else:
+            # 防脏数据：非 SQL 题不应携带 SQL 配置
+            data["sql_config"] = None
+
 
 class ProblemAPI(ProblemBase):
     @problem_permission_required