29 Commits
Author SHA1 Message Date
tao.chen b557fab2f2 update push message 2026-07-16 18:59:25 +08:00
tao.chen 37a1856ca7 update: bark message
Build Docker Image / Build image on tag (push) Successful in 2m44s
2026-07-16 18:49:21 +08:00
tao.chen 0e98878992 ci: update workflow 2026-07-16 18:44:05 +08:00
tao.chen 4659d43e8e fix: add path 2026-07-16 18:37:19 +08:00
tao.chen 3e4d095425 update: Dockerfile 2026-07-16 17:15:42 +08:00
tao.chen dcf978fe92 fix: mimo to mimocode 2026-07-16 16:54:50 +08:00
tao.chen 481bdf789a fix: mimo not found 2026-07-16 16:51:49 +08:00
tao.chen 15359e391e fix: ENTRYPOINT error 2026-07-16 16:47:42 +08:00
tao.chen 3336fdb5cd fix: build error 2026-07-16 16:42:17 +08:00
tao.chen 4a26c2bdfc fix: curl not found 2026-07-16 16:38:51 +08:00
tao.chen 689859881a update: Dockerfile 2026-07-16 16:24:30 +08:00
tao.chen 4028ef3dad update: add PATH 2026-07-16 15:44:07 +08:00
tao.chen 89297fb827 ci: remove buildx 2026-07-16 15:41:22 +08:00
tao.chen d608f491f1 ci: push bark 2026-07-16 15:08:46 +08:00
tao.chen 7ec98f4835 update: build.yml 2026-07-16 15:03:45 +08:00
tao.chen aa79292f6e ci: create workflow 2026-07-16 14:11:09 +08:00
tao.chen 165ab9bc29 udpate: Dockerfile
修复安装命令错误

Signed-off-by: tao.chen <1137626076@qq.com>
2026-07-16 11:03:38 +08:00
tao.chen 2871706c47 refactor: opencode to mimocode 2026-07-15 14:01:29 +08:00
tao.chen 59a1b5e4d4 update: Dockerfile 2026-07-03 16:07:16 +08:00
tao.chen 99447b2df6 feat: add csv2json scripts 2026-07-03 16:05:45 +08:00
tao.chen 0802ed634b feat: add pyspark script builder skill by wensichen 2026-07-01 18:38:43 +08:00
tao.chen 2f406bd64d update by wensicheng on 0629 2026-07-01 17:46:08 +08:00
tao.chen 0fc1c702d9 feat: spark mcp skill 2026-06-30 16:51:00 +08:00
tao.chen 43e5bbbf03 update: sql guard 2026-06-25 12:56:52 +08:00
tao.chen 87795131a3 fix bugs
1. "first keyword: --"  cosmetic bug
2. re error
2026-06-22 11:31:12 +08:00
tao.chen 6a6df9abf8 add sqlglot 2026-06-22 11:19:56 +08:00
tao.chen c8a5d5d792 add new skill;bash linux 2026-06-22 10:53:53 +08:00
tao.chen dd1496d669 update chinese skill 2026-06-22 10:36:30 +08:00
tao.chen f9b03384ab update, translate skill.md 2026-06-22 10:08:52 +08:00
19 changed files with 1788 additions and 1331 deletions
+72
View File
@@ -0,0 +1,72 @@
name: Build Docker Image
on:
push:
tags:
- 'v*'
workflow_dispatch:
jobs:
build:
name: Build image on tag
runs-on: host-shell
steps:
- name: Checkout
uses: https://gitea.com/actions/checkout@v4
# - name: Set up Docker Buildx
# uses: https://gitee.com/gitea-actions/setup-buildx-action@v3 # 推荐使用 Gitee/Gitea 国内镜像
- name: Login to Registry
uses: https://gitee.com/gitea-actions/login-action@v3
with:
registry: registry.cn-shanghai.aliyuncs.com
username: ${{ secrets.REGISTRY_USERNAME }}
password: ${{ secrets.REGISTRY_PASSWORD }}
- name: Show build context
run: |
echo "Ref: ${{ github.ref }}"
echo "Tag: ${GITHUB_REF_NAME}"
echo "Runner: $(uname -a)"
- name: Build Image
run: |
docker build -t opencode-custom:latest -t registry.cn-shanghai.aliyuncs.com/taochen/opencode-custom:latest .
- name: Verify image
run: |
docker images | grep opencode-custom
docker run --rm opencode-custom:latest --version
- name: Push Image
run: |
docker push registry.cn-shanghai.aliyuncs.com/taochen/opencode-custom:latest
- name: Notify release via Bark
if: always()
run: |
TITLE="${{ github.ref_name }}"
STATUS="${{ job.status }}"
if [ "${STATUS}" = "success" ]; then
EMOJI="✅"
else
EMOJI="❌"
fi
BODY="${EMOJI} opencode-custom:latest image build ${STATUS}\n branch ${{ github.ref_name }}\n workflow ${{ github.workflow }}\n run ${{ github.run_id }}"
PAYLOAD=$(cat <<EOF
{
"device_key": "6ZFVsxM95cuAjYoNLWSWaN",
"title": "${TITLE} - image build ${STATUS}",
"body": "${BODY}",
"group": "Gitea",
"isArchive": 1,
"ttl": 3600
}
EOF
)
curl -s \
-X POST http://101.43.40.124:81/push \
-H "Content-Type: application/json; charset=utf-8" \
-d "${PAYLOAD}"
+4
View File
@@ -115,3 +115,7 @@ The two non-pipeline skills (`logic-planner` taxonomy details, `sql-review` chec
- **Pipeline status changes** — never collapse gates without user confirmation. The pipeline's value comes from each stage being a separate, re-runnable hand-off; inlining them couples them.
- **Triggers in `description:` frontmatter** — these are what Claude Code matches against user prompts. Be specific (input contract + concrete trigger phrases). Vague triggers cause skill over-invocation.
- **Stale path references** — the bundled SKILL.md / scripts use `.claude/skills/...` paths from an earlier layout. When updating validator instructions, prefer the actual paths (`skills/...` on host, `/root/.config/opencode/skills/...` in image).
<!-- claude --resume a70186ab-eddf-442f-a7c9-6d3aa638f7f4 -->
+11 -5
View File
@@ -64,10 +64,16 @@ RUN uv pip install \
--index-url https://pypi.tuna.tsinghua.edu.cn/simple/ \
--requirements /tmp/requirements.txt
# 3. 在最终镜像中用正规的 npm 安装 opencode-ai,确保系统全局软链接生成完整
RUN npm install -g opencode-ai && npm cache clean --force
RUN opencode --version
COPY skills /root/.config/opencode/skills
RUN apt-get update && apt-get install -y --no-install-recommends \
curl \
ca-certificates \
&& rm -rf /var/lib/apt/lists/* \
&& curl -fsSL https://mimo.xiaomi.com/install | bash
ENTRYPOINT ["opencode"]
ENV PATH="/root/.mimocode/bin:${PATH}"
RUN mkdir -p /root/workspace
COPY skills /root/.config/mimocode/skills
COPY csv2json /root/workspace/csv2json
ENTRYPOINT ["mimo"]
Executable
+43
View File
@@ -0,0 +1,43 @@
#!/usr/bin/env python3
import csv
import json
import sys
import os
def main():
if len(sys.argv) < 2:
print("❌ 使用错误!")
print("用法: csv2json <输入CSV文件路径> [输出JSON文件路径]")
sys.exit(1)
csv_path = sys.argv[1]
json_path = sys.argv[2] if len(sys.argv) == 3 else os.path.splitext(csv_path)[0] + ".json"
try:
cleaned_data = []
with open(csv_path, mode='r', encoding='utf-8-sig') as f:
reader = csv.DictReader(f)
for row in reader:
# 💡 核心过滤逻辑:
# 只有当 key 不为 None、不为空字符串、且去掉空格后不为空时,才保留该列
filtered_row = {
key: value
for key, value in row.items()
if key is not None and str(key).strip() != ""
}
# 只有这一行过滤后还包含有效数据,才写入结果
if filtered_row:
cleaned_data = cleaned_data + [filtered_row]
with open(json_path, mode='w', encoding='utf-8') as f:
json.dump(cleaned_data, f, indent=4, ensure_ascii=False)
print(f"✅ 成功排除空列并转换: {csv_path} -> {json_path}")
except Exception as e:
print(f"❌ 转换失败: {e}")
if __name__ == '__main__':
main()
+1
View File
@@ -15,3 +15,4 @@ pymongo
redis
hdfs
pyhdfs
sqlglot
+88
View File
@@ -0,0 +1,88 @@
---
name: bash-linux
description: 当需要在 Linux/macOS Bash 环境或容器内查文件、搜文本、跑脚本、看日志、处理进程、使用管道、环境变量或编写小型 shell 命令时使用。它是辅助命令行规范,不属于 SQL 业务流水线。
---
# Bash Linux(命令行规范)
## 目标
让 agent 在 Bash/Linux/容器环境下用稳定、可解释、低风险的方式执行命令。先观察,再操作;先缩小范围,再修改。
## 基本原则
- 查文件优先用 `rg --files`,不可用时用 `find`
- 搜内容优先用 `rg -n`
- 读文件用 `sed -n``head``tail` 控制输出量。
- 脚本使用 `set -euo pipefail`
- 路径和变量要加引号。
- 删除、移动、改权限前必须确认目标范围。
- 长日志只摘关键部分,不整屏倾倒。
- 容器内的 `127.0.0.1` 指向容器自己,不等于宿主机。
## 常用命令
```bash
pwd
ls -la
find . -type f -name '*.py'
rg -n "keyword" .
sed -n '1,120p' file.txt
head -n 40 file.txt
tail -n 100 app.log
wc -l file.txt
```
## Docker/容器排查
```bash
docker ps
docker logs --tail 100 <container>
docker exec -it <container> sh
getent hosts host.docker.internal
curl -fsS http://host.docker.internal:8000/health
```
## 进程和端口
```bash
ps aux | grep '[p]ython'
lsof -i :8000
kill <pid>
```
## 脚本模板
```bash
#!/usr/bin/env bash
set -euo pipefail
main() {
if [ "$#" -lt 1 ]; then
echo "usage: $0 <arg>" >&2
exit 2
fi
echo "arg=$1"
}
main "$@"
```
## 输出报告
```yaml
command_report:
purpose: "为什么运行"
command: "命令摘要"
result: success | failed | skipped
important_output: "关键输出"
next_step: "下一步"
```
## 禁止行为
- 不要对未确认路径执行递归删除。
- 不要把不可信文件名直接拼进危险命令。
- 不要默认 GNU 命令参数在 macOS 上一定可用。
- 不要留下用户不需要的后台进程。
- 不要在没确认网络边界时把容器 localhost 当宿主机。
+123 -132
View File
@@ -1,158 +1,149 @@
---
name: logic-planner
description: Use when turning a validated data requirement (candidate tables/fields/joins already confirmed) into a step-by-step execution plan that a SQL generator can implement mechanically. Triggers include "拆解逻辑", "执行计划", "logic plan", "SQL 编排", "怎么算", "指标拆解", "join 顺序". Consumes the Validation Result from metadata-validator and produces a Logic Plan in a fixed step taxonomy (Source / Filter / Join / Transform / Aggregate / Output). Does NOT write SQL.
description: 当需求和元数据都已经确认,需要把业务逻辑拆成 SQL 生成前的确定性执行步骤时使用。它只规划 source、filter、join、dedupe、transform、window、aggregate、output,不写 SQL,不调用 MCP。
---
# Logic Planner
# Logic Planner(逻辑规划)
## Overview
## 目标
Decompose a validated data requirement into a deterministic pipeline of steps in a fixed taxonomy. The Logic Plan is the *only* input `sql-context-builder` needs to assemble a SQL Context — no further business reasoning is permitted downstream.
把已确认的 DRD 和元数据拆成可执行计算步骤,让后续 SQL 只是翻译计划,而不是重新猜业务逻辑。
Core principle: **decompose, don't translate.** This skill never writes SQL syntax. It only describes *what* must happen at each step in business/logical terms. The translation to SQL is a separate concern.
## 输入
## When to Use
Use when:
- `metadata-validator` returned status `VALIDATED` and you need to design how the query will run
- The user asks "how should I compute this", "what's the logic", "step-by-step plan" for a data requirement
- You need to disambiguate grain, metric formulas, deduplication, windowing, or ranking before SQL
Do NOT use when:
- `metadata-validator` returned `NEED_USER_CONFIRMATION` — go back and resolve open questions first
- The user has already provided a written execution plan and just needs SQL — go straight to `sql-context-builder`
## Inputs
1. **Validation Result** from `metadata-validator` (status `VALIDATED`)
2. The original business requirement (often embedded in `requirements-analysis` output)
3. `field_mapping` (mandatory) — every business field must be mapped to a physical `(table, column, type)` before planning starts
## Step Taxonomy
Every Logic Plan is a list of steps. Each step has exactly one `kind` from this set. Reuse the same kind for repeated operations (e.g. multiple joins, multiple filters).
| Kind | Purpose | Required fields |
|---|---|---|
| `source` | Read from a table or subquery | `alias`, `table`, `columns[]` |
| `filter` | Apply predicates to a named alias | `alias`, `predicates[]` |
| `join` | Combine two aliases on keys | `left`, `right`, `keys[]`, `type` (inner/left/right/full/semi/anti) |
| `transform` | Derive new columns (case-when, arithmetic, type cast) | `alias`, `expressions[]` |
| `aggregate` | Group and reduce | `alias`, `group_by[]`, `metrics[]` (each: `expr`, `alias`, `agg`) |
| `dedupe` | DISTINCT or row_number-based dedup | `alias`, `keys[]`, `strategy` |
| `window` | Ranking / running totals / lag-lead | `alias`, `partition_by[]`, `order_by[]`, `window_fn` |
| `output` | Final projection and ordering | `columns[]`, `order_by[]`, `limit?` |
Do not invent new kinds. If a transformation does not fit, surface it as `pending_question` and stop.
## Workflow
```dot
digraph logic_planner {
"Read Validation Result + field_mapping" [shape=box];
"Identify fact vs dimension tables & choose primary source" [shape=box];
"List filters (time range, status, business predicates)" [shape=box];
"Plan joins (alias A ON key = alias B.key)" [shape=box];
"Plan transforms (derived columns, type casts)" [shape=box];
"Plan dedupe / window if needed" [shape=box];
"Plan aggregate (grain + metrics)" [shape=box];
"Plan output projection + order" [shape=box];
"Any ambiguity?" [shape=diamond];
"Emit Logic Plan" [shape=box];
"Emit Logic Plan + pending_questions" [shape=box];
"Read Validation Result + field_mapping" -> "Identify fact vs dimension tables & choose primary source";
"Identify fact vs dimension tables & choose primary source" -> "List filters (time range, status, business predicates)";
"List filters (time range, status, business predicates)" -> "Plan joins (alias A ON key = alias B.key)";
"Plan joins (alias A ON key = alias B.key)" -> "Plan transforms (derived columns, type casts)";
"Plan transforms (derived columns, type casts)" -> "Plan dedupe / window if needed";
"Plan dedupe / window if needed" -> "Plan aggregate (grain + group_by)";
"Plan aggregate (grain + group_by)" -> "Plan output projection + order";
"Plan output projection + order" -> "Any ambiguity?";
"Any ambiguity?" -> "Emit Logic Plan" [label="no"];
"Any ambiguity?" -> "Emit Logic Plan + pending_questions" [label="yes"];
}
```yaml
requirements_output:
status: READY_FOR_METADATA
validation_result:
status: VALIDATED
field_mapping: {}
joins: []
```
## Required Reasoning (do not skip)
如果元数据没有 `VALIDATED`,停止并回到 `metadata-validator`
Before emitting the plan, you MUST explicitly state and record:
## 步骤类型
1. **Grain** — what does one output row represent? (e.g. one customer, one customer×month, one order)
2. **Metric formula** — for every metric, write the exact expression: numerator and denominator defined.
3. **Time grain** — confirm the time field and granularity (day, month, hour). State the window inclusively.
4. **Join cardinality** — for every join, predict 1:1 / 1:N / N:N. If N:N, mark it and ask the user whether to dedupe before joining.
5. **Dedup position** — if a fact table has row multiplicity (e.g. one customer with many addresses), decide *before* the join which side gets deduped and on which key.
6. **Null strategy** — for nullable join keys or measures, decide `INNER` vs `LEFT + COALESCE`.
只允许使用这些 `kind`
If any of the above is unclear, add a `pending_question` and stop. Do not guess.
- `source`:读取哪张表、哪些字段。
- `filter`:时间、分区、状态、业务条件过滤。
- `join`:多表关联。
- `dedupe`:去重,通常在 join 或 aggregate 前。
- `transform`:派生字段、分类、类型转换、标志位。
- `window`:首笔、末笔、排名、累计、lag/lead。
- `aggregate`:分组和指标计算。
- `output`:最终列、排序、limit 或写入意图。
## Output Contract
## 工作顺序
1. 定义输出粒度:一行代表什么。
2. 选事实表:承载核心事件或分母人群的表。
3. 列出 source:每张表只取必要字段。
4. 尽早放 filter:尤其是时间、分区和状态条件。
5. 固定时间窗口边界:下界包含、上界排除。
6. 规划 join:顺序、key、类型、基数假设。
7. 判断是否需要 dedupe/window:首登、首次注册、首笔成功事件等。
8. 规划 transform:标志位、事件顺序、空值处理。
9. 规划 aggregategroup by、指标公式、别名。
10. 规划 output:输出字段、排序、写入模式。
## 时间窗口硬规则
当需求为“近 N 天按 T-1 完整日”时,必须规划为:
```yaml
time_window:
lower_bound_sql: date_sub(current_date(), N)
upper_bound_sql: current_date()
boundary: left_closed_right_open
predicate_sql: event_time >= date_sub(current_date(), N) AND event_time < current_date()
```
禁止规划为 `BETWEEN ... AND date_add(current_date(), -1)`,尤其是 timestamp 字段。
## 转化/漏斗硬规则
转化类需求必须维护事件顺序。对“注册后成功付费”,推荐计划:
1. `first_login`:取窗口内每个用户的首次登录,形成分母。
2. `registered`:取用户注册事件,必要时去重到一个注册时间。
3. `qualified_payment`:在关联注册事件后筛选 `pay_status = 'SUCCESS' AND pay_time >= register_time`
4. `final_agg`:按维度计算分母、分子和转化率。
不要先对全量 `pay_order` 按用户取 `MIN(pay_time)` 再与 `register_time` 比较;这可能漏掉“历史早付费但注册后也付费”的用户。
## 必须显式写出的风险
- 1:N 或 N:N join 是否会放大事实行。
- 去重发生在 join 前还是 join 后。
- 指标是否需要 `count_distinct` 而不是 `count`
- 时间过滤是否能命中分区字段。
- left join 后维度缺失如何处理。
- 转化事件是否严格满足前后顺序。
- 分母为 0 时比率如何处理。
## 输出格式
```yaml
logic_plan:
grain: "<one sentence describing what one output row represents>"
fact_table: <table_name>
dimension_tables:
- <table_name>
status: PLANNED | NEED_USER_CONFIRMATION
grain: "结果一行代表什么"
fact_table: "事实表"
dimension_tables: []
time_window:
lower_bound_sql: "date_sub(current_date(), N)"
upper_bound_sql: "current_date()"
predicate_sql: "event_time >= date_sub(current_date(), N) AND event_time < current_date()"
boundary: left_closed_right_open
steps:
- kind: source
alias: o
table: loan_order
columns: [customer_id, loan_amount, apply_time, status]
- kind: filter
alias: o
predicates:
- "o.apply_time >= '<window_start>'"
- "o.apply_time < '<window_end>'"
- "o.status = 'SUCCESS'"
- kind: join
left: { alias: o }
right: { alias: c, table: customer_info }
keys: ["o.customer_id = c.customer_id"]
- id: S1
kind: source
name: first_login
table: "表名"
columns: ["字段"]
reason: "为什么需要"
- id: S2
kind: filter
target: "表名或中间结果"
predicates: ["业务过滤条件"]
partition_pruning: true
- id: S3
kind: join
left: "左表或CTE"
right: "右表或CTE"
keys: ["左字段 = 右字段"]
type: left
cardinality_assumption: 1:N # if known; else state "unknown"
- kind: transform
alias: o
expressions:
- expr: "CASE WHEN o.loan_amount >= 100000 THEN 'large' ELSE 'small' END"
alias: loan_bucket
- kind: aggregate
alias: o
group_by: ["c.city"]
cardinality: "1:1 | 1:N | N:1 | N:N | unknown"
row_growth_risk: low | medium | high
- id: S4
kind: aggregate
group_by: ["维度字段"]
metrics:
- expr: "SUM(o.loan_amount)"
alias: total_loan
agg: sum
- expr: "COUNT(DISTINCT o.customer_id)"
alias: customer_cnt
agg: count_distinct
- kind: output
columns: ["c.city AS city", "total_loan", "customer_cnt"]
order_by: ["total_loan DESC"]
limit: 100
pending_questions:
- "<ambiguity not yet resolved, blocking SQL generation>"
status: <PLANNED | NEED_USER_CONFIRMATION>
- name: "指标名"
agg: count_distinct | sum | avg | min | max | ratio
expression: "业务表达式"
alias: "指标别名"
- id: S5
kind: output
columns: ["输出列"]
order_by: []
write_intent: query_only | insert_into | insert_overwrite
assumptions: []
risks: []
pending_questions: []
```
## Forbidden Behaviors
## 用户可见摘要
| Rationalization | Reality |
|---|---|
| "I'll write the SELECT list directly, the user can read SQL" | Logic Plan uses business terms; SQL is downstream. |
| "Aggregation is obvious, skip the formula" | Explicit numerator/denominator avoids silent metric drift. |
| "1:N vs N:N doesn't matter for SELECT" | It matters: it determines dedup and inflation risk. |
| "Time grain is the user's problem" | Pick a grain. Wrong grain ⇒ wrong numbers. |
| "I'll add a HAVING clause at SQL time" | Logic Plan must list all filters, including post-aggregation. |
除了 YAML,必须输出这张表:
## Completion Criteria
```markdown
## 逻辑计划
| 步骤 | 中间结果 | 类型 | 说明 | 关键条件 |
|---|---|---|---|---|
```
Status `PLANNED` requires:
- Grain is one sentence, not a paragraph
- Every `field_mapping` entry is referenced by exactly one step
- Every join has a cardinality assumption
- Every metric has an explicit `agg` and an `alias`
- `pending_questions` is empty
## 停止条件
Otherwise: `NEED_USER_CONFIRMATION` and stop. Do not proceed to `sql-context-builder`.
如果粒度、指标公式、join 基数、去重策略、空值策略、时间归属或写入模式不明确,输出 `NEED_USER_CONFIRMATION`
+73 -137
View File
@@ -1,162 +1,98 @@
---
name: metadata-validator
description: Use when validating that a data requirement's candidate tables, fields, joins, and time columns actually exist in the project's metadata. Triggers include "validate metadata", "check if table exists", "field mapping", "join key validation", "确认表/字段", "校验 schema", "这个字段在哪张表", "口径对得上吗". Reads metadata files recursively across the workspace (JSON / Markdown / CSV / Excel / YAML / DDL / data dictionary) and produces a standardized Validation Result and Field Mapping.
description: 当需要确认候选表、字段、字段归属、字段类型、时间列、分区列和 join key 是否真实存在于工作区元数据中时使用。消费 requirements-analysis 的 requirements_output,产出 validation_result 和 field_mapping;不写 SQL,不调用 MCP。
---
# Metadata Validator
# Metadata Validator(元数据校验)
## Overview
## 目标
Bridge between the requirement's *expected* tables/fields and the *actual* schema living in workspace metadata. Output is consumed verbatim by `logic-planner` and `sql-context-builder` — they must NOT re-validate.
把需求里“想用的表和字段”与真实元数据对齐,防止模型凭感觉编表名、字段名和关联键。下游只能使用本 skill 确认过的字段映射。
Core principle: **never guess a table name, field name, or join relationship.** If a candidate is not provably present in metadata, surface it as a `pending_question` and stop at status `NEED_USER_CONFIRMATION`.
## 输入
## When to Use
Use when:
- `requirements-analysis` has produced candidate tables/fields and you need to confirm they exist in real metadata
- The user asks "does this table/field exist", "which file describes table X", "map business field to physical column"
- You need to detect join keys, time columns, or aggregation-suitable numeric columns
- The user is about to write SQL and you must guard against hallucinated column names
Do NOT use when:
- The user has already supplied exact table+column names and confirmed them — go straight to `logic-planner`
- The task is code generation without any data warehouse context
## Inputs
1. From `requirements-analysis`:
- `candidate_tables`: list of table names mentioned or implied
- `candidate_fields`: list of business field names with their proposed table
- `business_logic`: short summary of what is being computed
2. Workspace metadata: any file under the working directory matching metadata conventions (see below)
## Workflow
```dot
digraph metadata_validator {
"Receive candidate tables/fields" [shape=box];
"Recursively scan workspace for metadata files" [shape=box];
"Parse each file into canonical (table, field, type) records" [shape=box];
"Check candidate tables exist" [shape=box];
"Check candidate fields exist in their table" [shape=box];
"Fuzzy match missing fields against parsed metadata" [shape=box];
"Detect join keys & time columns" [shape=box];
"All checks pass?" [shape=diamond];
"Build Field Mapping" [shape=box];
"Emit Validation Result (VALIDATED)" [shape=box];
"Emit Validation Result (NEED_USER_CONFIRMATION)" [shape=box];
"Receive candidate tables/fields" -> "Recursively scan workspace for metadata files";
"Recursively scan workspace for metadata files" -> "Parse each file into canonical (table, field, type) records";
"Parse each file into canonical (table, field, type) records" -> "Check candidate tables exist";
"Check candidate tables exist" -> "Check candidate fields exist in their table";
"Check candidate fields exist in their table" -> "Fuzzy match missing fields against parsed metadata";
"Fuzzy match missing fields against parsed metadata" -> "Detect join keys & time columns";
"Detect join keys & time columns" -> "All checks pass?";
"All checks pass?" -> "Build Field Mapping" [label="yes"];
"All checks pass?" -> "Emit Validation Result (NEED_USER_CONFIRMATION)" [label="no"];
"Build Field Mapping" -> "Emit Validation Result (VALIDATED)";
}
```yaml
requirements_output:
status: READY_FOR_METADATA
candidate_tables: []
candidate_fields: []
join_hints: []
metadata_files: "当前工作区中的数据字典、DDL、schema、CSV 表头、Markdown 表格、Excel 等"
```
## Metadata Discovery
如果 `requirements_output.status` 不是 `READY_FOR_METADATA`,停止并回到 `requirements-analysis`
- Recursively scan the **current working directory** (and any subdirectories) — do not assume a `metadata/` folder exists.
- Filenames often resemble table names (e.g. `loan_order.json`, `customer_info.md`, `dim_product.csv`) but **never trust the filename alone** — open the file and confirm the table identifier inside.
- Accepted formats (auto-detect by content, not extension):
- JSON / JSONL
- YAML
- Markdown tables / headings
- CSV (with header)
- Excel (`.xlsx`, `.xls`) — use `openpyxl` or `pandas.read_excel`
- DDL / `CREATE TABLE` statements
- Free-form data dictionaries (parse key-value lines or `field | type | comment` tables)
- Convert every file into the canonical record shape before validating:
## 元数据查找顺序
```yaml
tables:
- table: <physical_table_name>
source_file: <relative_path>
fields:
- name: <field_name>
type: <string|int|long|double|decimal|date|timestamp|boolean|...>
comment: <optional>
```
1. 递归查看工作区中的元数据文件,不只相信文件名。
2. 优先读取结构化文件:DDL、JSON/YAML schema、CSV/Excel 数据字典、Markdown 表格。
3. 对每张表记录证据:文件路径、表名出现位置、字段名、类型、注释、枚举。
4. 文件名只能作为线索,不能作为存在性证据。
## Validation Rules
## 校验规则
Run **all** of the following. Any failure ⇒ status `NEED_USER_CONFIRMATION`.
- 字段在别的表里存在,不等于在目标表里存在。
- 字段名相似不等于字段可替换,必须保留候选并让用户确认。
- 时间字段有多个时必须问清楚事件时间、分区时间和统计归属。
- join key 有多个候选时必须问清楚,并标注基数风险。
- 金额、数量、比率类指标要确认类型、单位和精度。
- 结果表如果要写入,必须校验目标表存在性和字段兼容性;`INSERT OVERWRITE` 还需要用户显式确认。
1. **Table existence** — every candidate table must appear in the parsed metadata.
2. **Field existence** — every candidate field must appear in its candidate table.
3. **Field ownership** — a field may exist but belong to a different table; never silently swap.
4. **Type sanity** — aggregation targets (sum / avg / count) must be numeric; time filters must be date/timestamp/string-encoded-date.
5. **Necessary fields** — flag if common fields are missing for the business logic (stat metric, time, join key, dimension).
6. **Fuzzy match** — for every missing field, score candidates by name similarity (`customer_id` ↔ `cust_id`, `cust_no`, `customer_no`) and Levenshtein distance; surface top-N.
7. **Join detection** — across the validated tables, look for shared `*_id`, `*_no`, `*_code` columns. If multiple plausible join keys exist, ask.
8. **Time column disambiguation** — list every time-like field per table (`create_time`, `apply_time`, `txn_date`, `update_time`, `dt`); ask which one defines the time grain.
9. **Coverage** — verify that dimensions, metrics, filters, and sort keys are all present.
## 工作顺序
## Forbidden Behaviors
1. 整理所有候选表和候选字段。
2. 从元数据中确认表是否存在。
3. 校验字段是否存在于正确表中,并记录类型和来源。
4. 校验指标字段是否适合聚合,时间字段是否适合时间过滤。
5. 校验 join key 两边都存在,记录类型一致性和基数假设。
6. 校验分区字段、状态枚举、金额单位等会影响结果的细节。
7. 对缺失项给出证据和候选替代,不猜。
8. 如果缺表、缺字段或 join/time 不明确,停止。
| Rationalization | Reality |
|---|---|
| "The filename is `loan_order.json`, must be that table" | Filename is a hint. Parse the file to confirm. |
| "Field exists somewhere, so the join is fine" | Wrong table ownership invalidates the join. Always check ownership. |
| "The user probably means `cust_no`, just use it" | Surface as a candidate and ASK. Never auto-substitute. |
| "I'll guess the time field" | Multiple time fields ⇒ ASK. The grain defines the entire query. |
| "This metadata is enough, no need to scan further" | Always scan recursively. One file can describe many tables. |
## Output Contract
The output is a single YAML document. Downstream skills consume it as-is.
## 输出格式
```yaml
validation_result:
validated_tables:
- <table_name>
validated_fields:
- <table_name>.<field_name>
missing_tables: []
missing_fields:
- <field_name_not_found_anywhere>
candidate_fields:
<missing_field_name>:
- <candidate_1>
- <candidate_2>
candidate_tables:
<missing_table_name>:
- <candidate_1>
join_candidates:
- left: <table_a>.<field>
right: <table_b>.<field>
confidence: <high|medium|low>
time_field_candidates:
<table_name>:
- <field>
- <field>
status: VALIDATED | NEED_USER_CONFIRMATION
metadata_sources:
- <relative_path_to_metadata_file>
pending_questions:
- "<human-readable question, ideally with options A/B/C>"
- path: "元数据文件路径"
evidence: "表/字段证据摘要"
validated_tables:
- table: "物理表名"
role: fact | dimension | lookup | target | unknown
source_file: "元数据来源"
validated_fields:
- table: "表名"
column: "字段名"
type: "字段类型"
comment: "字段注释"
source_file: "元数据来源"
field_mapping:
<business_field_name>:
table: <physical_table>
column: <physical_field>
type: <data_type>
status: <VALIDATED | NEED_USER_CONFIRMATION>
"业务字段名":
table: "物理表名"
column: "物理字段名"
type: "字段类型"
role: dimension | metric | filter | time | partition | join_key | output
source_file: "元数据来源"
confidence: high | medium | low
joins:
- left: "表A.字段"
right: "表B.字段"
type: inner | left | right | full | unknown
cardinality: "1:1 | 1:N | N:1 | N:N | unknown"
confidence: high | medium | low
time_fields:
- table: "表名"
event_time: "事件时间字段"
partition_time: "分区字段,可为空"
missing_tables: []
missing_fields: []
alternatives:
"缺失项": ["可能候选"]
pending_questions:
- "需要用户确认的问题"
```
`field_mapping` is the **single source of truth** for downstream `logic-planner` and `sql-context-builder`. If a business field has no mapping, the request cannot proceed — add it to `pending_questions`.
## 停止条件
## Completion Criteria
Status `VALIDATED` is only allowed when **all** of the following hold:
- Every candidate table is present in `validated_tables`
- Every candidate field has a `field_mapping` entry
- All join keys are decided (or trivially obvious with `high` confidence)
- The time column for the query grain is decided
- `pending_questions` is empty
Anything else ⇒ `NEED_USER_CONFIRMATION` and stop. Do not proceed to `logic-planner`.
只在所有必需表、字段、join key、时间字段和目标写入字段都确认后输出 `VALIDATED`。否则输出 `NEED_USER_CONFIRMATION`
+47
View File
@@ -0,0 +1,47 @@
---
name: pyspark-script-builder
description: Use after sql-review to wrap reviewed_sql into a complete PySpark script. Inputs are reviewed_sql, app_name, and script_name; ask the user for app_name or script_name if missing; preserve reviewed_sql business logic and output app_name, script_name, and pyspark_code only.
---
# PySpark Script Builder
Use this skill only after `sql-review`, when `reviewed_sql` is already available and the user needs it wrapped as a complete PySpark script.
## Required Inputs
- `reviewed_sql`: SQL text that has already been reviewed.
- `app_name`: Spark application name. If missing, ask the user before producing output. Do not invent or derive it.
- `script_name`: Script file name. If missing, ask the user before producing output. Do not invent or derive it.
## Rules
- Preserve the business logic of `reviewed_sql`. Do not rewrite tables, columns, predicates, joins, aggregations, ordering, limits, comments, or SQL semantics.
- Only wrap `reviewed_sql` in the fixed PySpark script structure below.
- Do not add file persistence, job creation, confirmation, cluster launch, queue/resource selection, monitoring, logs, or lifecycle steps.
- Escape `app_name` only as needed to keep a valid Python double-quoted string.
- Put `reviewed_sql` inside a raw triple-quoted string exactly in the `sql = r"""..."""` assignment.
- Do not run the SQL or validate it again; this skill only builds script text.
## Output
Return only these fields:
- `app_name`
- `script_name`
- `pyspark_code`
Build `pyspark_code` with this exact structure:
```python
from pyspark.sql import SparkSession
APP_NAME = "<app_name>"
def main():
spark = SparkSession.builder.appName(APP_NAME).enableHiveSupport().getOrCreate()
sql = r"""<reviewed_sql>"""
spark.sql(sql)
spark.stop()
if __name__ == "__main__": main()
```
+54 -332
View File
@@ -1,363 +1,85 @@
---
name: pyspark-sql-guardrails
description: Use when writing or reviewing PySpark scripts, Spark SQL strings, spark.sql calls, SQL loaded from config files, ETL jobs, warehouse maintenance scripts, or data pipeline code where SQL must be limited to SELECT and INSERT only.
description: 当生成、展示、评审或执行 PySpark/Spark SQL 字符串时使用。它负责将 SQL 限制在 SELECT、WITH...SELECT、INSERT...SELECT 安全白名单内,并阻止危险 DDL/DML/维护语句、堆叠语句和未确认的 INSERT OVERWRITE。
---
# PySpark SQL Guardrails
# PySpark SQL GuardrailsSQL 安全闸门)
## Validation Gate (Read This First — Non-Negotiable)
## 目标
**Every SQL string this skill produces is UNTRUSTED until it has been run through the bundled validator. The validator runs in one command, immediately after the SQL is generated, and BEFORE the SQL is shown to the user. There is no path that skips this step.**
在 SQL 被展示、写入 PySpark 脚本或交给 `spark.sql` 前,先通过白名单校验。通过本 skill 只代表 SQL 语句类型安全,不代表业务逻辑正确。
The whole flow is two commands and one condition:
## 允许
```bash
# Step 1 — Generate the SQL (do this in your head / draft)
# Step 2 — Run the validator on it (do this IMMEDIATELY, in the same turn)
echo "<the SQL you just wrote>" | python3 .claude/skills/pyspark-sql-guardrails/scripts/validate_sql.py
# Step 3 — Read the output:
# PASS - first keyword: select, body length: N → you may now show the SQL
# FAIL - <reason> → STOP. Fix the SQL. Re-run.
```
**If you have not pasted the `PASS -` line into the conversation yet, you are not allowed to paste the SQL.** The validator is the gate. The SQL sits on the other side. Do not pass the gate empty-handed.
This rule is enforced by the skill, not by an external system. Honoring it is a precondition for any output that contains a SQL string.
---
## Overview
PySpark SQL in this environment is allowlist-only: generated or executed SQL may be `SELECT` or `INSERT` only. Treat every SQL string, formatted SQL template, config-loaded SQL, notebook cell, and `spark.sql(...)` call as unsafe until validated.
Core principle: **never execute SQL by checking that it is “not obviously dangerous”; execute only after proving it is one allowed statement whose first real keyword is SELECT or INSERT.**
## Mandatory Validation (Hard Rule)
**Every SQL string MUST pass through `assert_select_or_insert()` before it can reach `spark.sql(...)` — no exceptions.** This is non-negotiable and applies to:
- SQL you just generated (literal, f-string, `.format(...)`, `+` concatenation)
- SQL loaded from YAML / JSON / INI / env / CLI / config files
- SQL passed in via parameters, function arguments, or notebook variables
- SQL pasted in from a chat message
- SQL produced by a templating engine (Jinja, string.Template, etc.)
The only acceptable call shapes:
```python
# Preferred: wrapper that validates + executes atomically
safe_spark_sql(spark, sql)
# Or: validate first, then execute explicitly
assert_select_or_insert(sql) # raises on violation
spark.sql(sql)
```
**Forbidden call shapes (any of these ⇒ pipeline failure):**
```python
spark.sql(sql) # direct, no validation
spark.sql(f"SELECT ... {user_input} ...") # f-string into spark.sql
spark.sql(config["sql"]) # config-driven without validation
spark.sql(open("queries/xxx.sql").read()) # file-loaded without validation
```
If `assert_select_or_insert()` raises, the pipeline stops. Do not weaken the rule, do not strip forbidden keywords with regex, do not "rewrite" the SQL into a safe-looking form. Reject and re-author.
## Required Rule
Allowed:
- `SELECT ...`
- `WITH ... SELECT ...`
- `INSERT INTO ... SELECT ...`
- `INSERT OVERWRITE ... SELECT ...` only if the user explicitly allows overwrite for this job; otherwise ask before using it.
- `INSERT OVERWRITE ... SELECT ...`,但必须拿到用户明确允许 overwrite 的确认,并使用 `allow_overwrite=true`
Forbidden, even for maintenance or metadata refresh:
- `DELETE`, `TRUNCATE`, `DROP`, `ALTER`, `CREATE`, `REPLACE`, `MERGE`, `UPDATE`
- `MSCK REPAIR`, `REFRESH`, `ANALYZE`, `CACHE`, `UNCACHE`, `VACUUM`, `OPTIMIZE`
- Multiple statements separated by semicolons
- Arbitrary SQL from YAML/JSON/env/CLI passed directly to `spark.sql`
## 禁止
**Execution gate:** `assert_select_or_insert(sql)` (or `safe_spark_sql(spark, sql)`) MUST be called for **every** `spark.sql` invocation. There is no fast path. See *Mandatory Validation (Hard Rule)* above.
- `DROP``DELETE``UPDATE``MERGE``ALTER``CREATE``REPLACE`
- `TRUNCATE``MSCK``REFRESH``ANALYZE``CACHE``UNCACHE`
- `VACUUM``OPTIMIZE``CALL``GRANT``REVOKE`
- 用分号堆叠多条语句
- `INSERT ... VALUES` 或不基于 `SELECT` 的写入
- 未校验就直接进入 `spark.sql(sql)`
## Quick Reference
## 使用脚本
| Situation | Do |
|---|---|
| Need rows | `safe_spark_sql(spark, "SELECT ...")` |
| Need to write rows | `safe_spark_sql(spark, "INSERT INTO table SELECT ...")` |
| Need cleanup/dedup/delete | Use SELECT to derive clean data, then `safe_spark_sql` with INSERT into an approved target/staging table; do not delete old data |
| Need schema/table/partition maintenance | Stop and ask; do not emit DDL or metadata SQL |
| SQL comes from config | `assert_select_or_insert(sql)` first, **then** `spark.sql(sql)` |
| User asks “quick fix” | Safety rule still applies — run `assert_select_or_insert` first |
| Generated SQL from `sql-context-builder` | Always run `assert_select_or_insert` before executing |
## Safe Pattern
**MUST:** every `spark.sql` call site uses `safe_spark_sql` (or calls `assert_select_or_insert` immediately before `spark.sql`). The validator is the only thing standing between a generated string and the cluster.
Use validation at the boundary where SQL enters execution:
```python
import re
_FORBIDDEN_SQL = re.compile(
r"\b(delete|truncate|drop|alter|create|replace|merge|update|msck|refresh|analyze|cache|uncache|vacuum|optimize)\b",
re.IGNORECASE,
)
_LEADING_COMMENTS = re.compile(r"\A\s*(?:--[^\n]*\n|/\*.*?\*/\s*)*", re.DOTALL)
def assert_select_or_insert(sql: str) -> str:
text = sql.strip()
if not text:
raise ValueError("SQL is empty")
# Allow one optional trailing semicolon, but reject stacked statements.
body = text[:-1].strip() if text.endswith(";") else text
if ";" in body:
raise ValueError("Multiple SQL statements are not allowed")
normalized = _LEADING_COMMENTS.sub("", body).lstrip()
first = normalized.split(None, 1)[0].lower() if normalized else ""
if first not in {"select", "with", "insert"}:
raise ValueError(f"Only SELECT and INSERT SQL are allowed, got {first!r}")
if _FORBIDDEN_SQL.search(normalized):
raise ValueError("Forbidden SQL keyword found")
if first == "with" and not re.search(r"\bselect\b", normalized, re.IGNORECASE):
raise ValueError("WITH statements must be SELECT queries")
return body
def safe_spark_sql(spark, sql: str):
return spark.sql(assert_select_or_insert(sql))
# Good
safe_spark_sql(spark, """
INSERT INTO analytics.daily_customer_snapshot
SELECT * FROM staging.daily_customer_snapshot
""")
# Bad: raises before Spark sees it
safe_spark_sql(spark, "ALTER TABLE analytics.daily_customer_snapshot DROP PARTITION (dt='2026-06-10')")
```
If the project already has a SQL parser such as `sqlglot`, prefer AST validation over regex. Still keep the same allowlist: one statement, top-level SELECT or INSERT, no forbidden DDL/DML/maintenance commands anywhere.
## Local Verification Scaffold (Run After Every Generation)
The validator is useless if it is never executed. After **every** SQL string is produced (whether by you, a sub-agent, a template, or a tool), the next action is to run it through `assert_select_or_insert` in Python and only then show the result to the user.
This skill ships its own validator and test set inside the skill directory so the contract is self-contained and versioned with the skill itself. **Do not redefine the regex inline at call sites. Always import from the bundled file.**
```
.claude/skills/pyspark-sql-guardrails/
├── SKILL.md ← this file
└── scripts/
├── sql_guard.py ← canonical validator (assert_select_or_insert + safe_spark_sql)
├── validate_sql.py ← one-line CLI: pipe SQL in, get PASS/FAIL out
└── test_sql_guard.py ← 18-case standard test set
```
### Step 1 — Use the bundled validator (no copy-pasting, no rewriting)
The skill ships a one-line CLI at `validate_sql.py`. Use it. Do not write a different invocation, do not paste inline regex, do not call `assert_select_or_insert` directly from a sub-agent unless the bundled CLI is also broken. The CLI is a thin wrapper around `assert_select_or_insert`; the function it calls is the source of truth.
**Two ways to feed SQL into the validator:**
```bash
# Way A (preferred for multi-line): pipe via stdin
cat <<'EOF' | python3 .claude/skills/pyspark-sql-guardrails/scripts/validate_sql.py
SELECT
c.city AS city,
SUM(o.loan_amount) AS total_loan
FROM loan_order o
INNER JOIN customer_info c ON o.customer_id = c.customer_id
WHERE o.status = 'SUCCESS'
GROUP BY c.city
EOF
# Way B (single-line only): pass as first argument
python3 .claude/skills/pyspark-sql-guardrails/scripts/validate_sql.py "SELECT 1 FROM dual"
```
**Exit codes:**
- `0``PASS - first keyword: <verb>, body length: <N>`
- `1``FAIL - <reason>` (the reason names the rule that was violated)
- `2` → usage error (no SQL provided)
The bundled `sql_guard.py` and `validate_sql.py` together:
```python
# sql_guard.py (bundled)
import re
_FORBIDDEN_SQL = re.compile(
r"\b(delete|truncate|drop|alter|create|replace|merge|update|msck|refresh|analyze|cache|uncache|vacuum|optimize)\b",
re.IGNORECASE,
)
_LEADING_COMMENTS = re.compile(r"\A\s*(?:--[^\n]*\n|/\*.*?\*/\s*)*", re.DOTALL)
def assert_select_or_insert(sql: str) -> str:
text = sql.strip()
if not text:
raise ValueError("SQL is empty")
body = text[:-1].strip() if text.endswith(";") else text
if ";" in body:
raise ValueError("Multiple SQL statements are not allowed")
normalized = _LEADING_COMMENTS.sub("", body).lstrip()
first = normalized.split(None, 1)[0].lower() if normalized else ""
if first not in {"select", "with", "insert"}:
raise ValueError(f"Only SELECT and INSERT SQL are allowed, got {first!r}")
if _FORBIDDEN_SQL.search(normalized):
raise ValueError("Forbidden SQL keyword found")
if first == "with" and not re.search(r"\bselect\b", normalized, re.IGNORECASE):
raise ValueError("WITH statements must be SELECT queries")
return body
def safe_spark_sql(spark, sql: str):
return spark.sql(assert_select_or_insert(sql))
# validate_sql.py (bundled, abridged)
import sys, os
SKILL_DIR = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, SKILL_DIR)
from sql_guard import assert_select_or_insert
def main():
sql = sys.argv[1] if len(sys.argv) > 1 else sys.stdin.read()
try:
body = assert_select_or_insert(sql)
except ValueError as e:
print(f"FAIL - {e}"); return 1
first = body.split(None, 1)[0].lower()
print(f"PASS - first keyword: {first}, body length: {len(body)}")
return 0
```
### Step 2 — Required call pattern after every SQL generation
The same command, every time, with the SQL you just produced. No variation, no shortcut, no "I'll run it later":
```bash
echo "<THE_SQL_YOU_JUST_GENERATED>" | python3 .claude/skills/pyspark-sql-guardrails/scripts/validate_sql.py
```
**Show the validator's output to the user before showing the SQL.** The expected reply shape:
本 skill 自带脚本:
```text
PASS - first keyword: select, body length: 420
scripts/sql_guard.py 校验库
scripts/validate_sql.py 命令行入口
scripts/test_sql_guard.py 标准测试集
```
If you see `FAIL - <reason>` instead, the SQL is not allowed past the gate. Surface the failure to the user verbatim, fix the SQL, re-run the validator. Loop until `PASS`.
### Step 3 — Standard test set (run on any change to the validator)
The skill bundles `test_sql_guard.py` with 6 EXPECT_OK + 12 EXPECT_RAISE cases. Run it from the skill directory any time the validator changes, and as part of CI:
校验一条 SQL
```bash
# From project root
python3 .claude/skills/pyspark-sql-guardrails/scripts/test_sql_guard.py
# Or from the scripts/ directory
cd .claude/skills/pyspark-sql-guardrails/scripts
python3 test_sql_guard.py
python scripts/validate_sql.py "SELECT 1"
```
Expected output: 6 `[OK] EXPECT_OK` lines, 12 `[OK] EXPECT_RAISE: raised as expected` lines, terminated with `ALL TESTS PASSED`. Any deviation is a regression in the validator — fix the validator, not the tests.
从文件校验:
### Step 4 — Failure protocol
When the validator raises (in CLI form, the `FAIL - <reason>` line):
1. **Stop the pipeline.** Do not continue to the next stage.
2. **Read the reason.** It names the rule that was violated (empty / multi-statement / wrong verb / forbidden keyword / WITH without SELECT).
3. **Do not patch the validator to accept the SQL.** The validator is the source of truth; the SQL is wrong.
4. **Do not strip forbidden keywords with regex** to "clean up" the SQL. Reject and re-author the SQL itself.
5. **Re-run the validator** with the corrected SQL. Loop until `PASS`.
### Forbidden Behaviors (validator execution)
| Rationalization | Reality |
|---|---|
| "I'll just `spark.sql(sql)` and trust it" | That is what the guardrail is preventing. Run the validator. |
| "The validator output is in the previous turn" | Validators are stateful only via the function call. Re-run it. |
| "It's a one-liner, eyeballing is enough" | Eyeballing is exactly the failure mode this guardrail exists to prevent. |
| "I already ran it on a similar SQL" | Each SQL string is a new value. Re-run on the actual string being shipped. |
| "I'll inline a regex instead of using `validate_sql.py`" | The skill ships one canonical CLI. Inline copies drift and miss updates. |
| "The SQL is in a code block, not yet 'executed'" | Showing the SQL to the user IS the execution. The gate is before the code block, not after. |
## Common Rationalizations
| Excuse | Reality |
|---|---|
| “It is just metadata refresh.” | `REFRESH`, `MSCK`, and `ANALYZE` are outside SELECT/INSERT. Ask first. |
| “MERGE is non-destructive.” | Policy allows SELECT/INSERT only. `MERGE` is forbidden. |
| “DELETE+INSERT is the existing pattern.” | Existing unsafe patterns do not override the rule. |
| “The config is trusted.” | Config SQL still needs allowlist validation before `spark.sql`. |
| “I can strip forbidden words with regex.” | Do not rewrite unsafe SQL into safe-looking SQL. Reject it. |
| “We need cleanup before insert.” | Build cleaned data with SELECT and write with INSERT; do not mutate/delete existing data. |
| **“The SQL is hard-coded, no need to validate.”** | **Hard-coded strings still must pass through `assert_select_or_insert`. The function checks the *string*, not its provenance. A literal in source code is no safer than a config value.** |
| **“I already eyeballed it, the verb is SELECT.”** | **Eyeballing is exactly the failure mode this guardrail prevents. Trust the function, not the eye. If `assert_select_or_insert` was not called, the SQL is untrusted by definition.** |
| **“I'll wrap it later / in a follow-up PR.”** | **No. Add the wrapper at the call site *now*. A naked `spark.sql(...)` in committed code is a regression, not a TODO.** |
## Red Flags
Stop and add validation or ask the user when you see:
- `spark.sql(config["sql"])`, YAML SQL, env SQL, CLI SQL, or f-string SQL from external input
- Any SQL verb other than SELECT/WITH/INSERT
- `INSERT OVERWRITE` without explicit approval for overwrite semantics
- Semicolon-separated SQL batches
- Partition repair, schema migration, stale partition cleanup, compaction, vacuum, or stats collection requests
- Requests phrased as “quick cleanup”, “just refresh metadata”, “drop old partitions”, or “reuse DELETE+INSERT”
- **A direct `spark.sql(sql)` call with no preceding `assert_select_or_insert(sql)` (or no use of `safe_spark_sql`) — even if the SQL string is a hard-coded literal**
- **A pull request, diff, or code review that introduces a new `spark.sql(...)` site without the wrapper**
## Common Mistakes
- Checking only for `DELETE` and `TRUNCATE`; Spark risk also includes `DROP`, `ALTER`, `MERGE`, `UPDATE`, `MSCK`, `REFRESH`, `ANALYZE`, `VACUUM`, and `OPTIMIZE`.
- Validating generated SQL but not config-loaded SQL.
- Allowing multiple statements because the first statement is SELECT.
- Treating comments as harmless while a later statement contains forbidden SQL.
- Using `CREATE OR REPLACE TEMP VIEW` through SQL; prefer DataFrame `createOrReplaceTempView` if a temp view is needed.
- **Calling `spark.sql(sql)` directly without routing through `assert_select_or_insert` / `safe_spark_sql`. The allowlist is enforced at the call site, not at SQL-generation time. A generated SQL that *looks* safe is not *validated* safe until the function has run on it.**
- **Writing inline `if sql.startswith("SELECT"): spark.sql(sql)` checks. That is not validation. Only `assert_select_or_insert` (or an equivalent AST parser with the same allowlist) counts.**
---
## Pipeline Position
This skill is the **write-time / execution boundary** of the PySpark SQL pipeline:
```
requirements-analysis → metadata-validator → logic-planner → sql-context-builder → pyspark-sql-guardrails → sql-review
```bash
python scripts/validate_sql.py --file query.sql
```
**Required upstream skill:** `sql-context-builder` — every generated SQL must come with a `sql_context` that froze aliases, join keys, and field sources. SQL without an associated context is a `HIGH` finding under `sql-review` (see check #1: reference integrity).
JSON 输出:
**Required downstream skill:** `sql-review` — this guardrail validates *what statements are allowed*; `sql-review` validates *whether the allowed statement is correct*. Both must pass before `spark.sql(...)` is called. Do not skip `sql-review` because the guardrail passed.
```bash
python scripts/validate_sql.py --json "SELECT 1"
```
**Execution gate (enforced by this skill):** the only path from a SQL string to the cluster is `assert_select_or_insert(sql)` (or its wrapper `safe_spark_sql(spark, sql)`). A direct `spark.sql(sql)` is a pipeline violation. This rule overrides the pipeline ordering — even if `sql-review` has already passed, the SQL still has to pass through this guardrail at execution time. Guardrail and reviewer are independent checks; one does not satisfy the other.
允许 overwrite 时:
**The guardrail does not check:**
- Whether the column actually exists in the table (that is `sql-review`'s job via the SQL Context)
- Whether the join key is correct (that is `sql-review`'s job)
- Whether aggregation is duplicated or rows will inflate (that is `sql-review`'s job)
- Whether time/partition filters are present (that is `sql-review`'s job)
```bash
python scripts/validate_sql.py --allow-overwrite "INSERT OVERWRITE target SELECT * FROM source"
```
**The guardrail does check:**
- First real keyword is `SELECT`, `WITH ... SELECT`, or `INSERT ... SELECT`
- No stacked statements (`;`-separated batches)
- No forbidden verbs anywhere in the body
- `INSERT OVERWRITE` is gated on explicit user approval
## 输出格式
When the guardrail raises, fix the SQL, do not weaken the guardrail. When in doubt, ask the user before generating `INSERT OVERWRITE` or any DDL.
```yaml
sql_guard:
status: PASS | FAIL
first_keyword: select | with | insert | unknown
reason: "失败原因,PASS 时为空"
```
## 必须执行的关口
1. SQL 生成后、展示给用户前,先运行 guard。
2. PySpark 代码中出现 `spark.sql(...)` 时,SQL 字符串必须先通过 guard。
3. guard 失败时,回到 SQL 生成步骤修 SQL,不要削弱校验器。
4. guard 通过后仍然必须进入 `sql-review`
## 规则
- 校验必须针对最终要展示或执行的那条 SQL。
- 肉眼看过不算通过。
- 一条类似 SQL 通过,不代表当前 SQL 通过。
- 字符串字面量和注释里的危险词不应误报。
- 反引号字段名里的危险词不应误报,但不建议这样命名。
- `INSERT OVERWRITE` 默认失败,除非用户明确确认 overwrite。
@@ -1,52 +1,137 @@
"""PySpark SQL Guardrails — validator and safe execution wrapper.
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Dependency-free Spark SQL allowlist guard.
This module is the canonical implementation referenced by
.claude/skills/pyspark-sql-guardrails/SKILL.md. Do not redefine the
regex inline at call sites; do not write "lighter" versions. Always
import this module and route `spark.sql(...)` through `safe_spark_sql`.
The validator is pure-Python and has no Spark / I/O dependencies. It
is safe to import in unit tests, CI, notebooks, and ad-hoc scripts.
The guard answers one question only: is this SQL statement type safe enough
to show or pass to spark.sql? It does not prove business correctness.
"""
from __future__ import annotations
import re
from dataclasses import dataclass
_FORBIDDEN_SQL = re.compile(
r"\b(delete|truncate|drop|alter|create|replace|merge|update|msck|refresh|analyze|cache|uncache|vacuum|optimize)\b",
re.IGNORECASE,
)
_LEADING_COMMENTS = re.compile(r"\A\s*(?:--[^\n]*\n|/\*.*?\*/\s*)*", re.DOTALL)
ALLOWED_FIRST = {"select", "with", "insert"}
FORBIDDEN = {
"delete", "truncate", "drop", "alter", "create", "replace", "merge", "update",
"msck", "refresh", "analyze", "cache", "uncache", "vacuum", "optimize",
"call", "grant", "revoke",
}
def assert_select_or_insert(sql: str) -> str:
"""Return the SQL body if it is a single SELECT/INSERT, else raise ValueError.
@dataclass(frozen=True)
class GuardResult:
status: str
first_keyword: str
reason: str = ""
Hard-allowlist. No I/O, no logging side effects, no Spark dependency.
Safe to call in unit tests, CI, notebooks, and ad-hoc scripts.
"""
text = sql.strip()
if not text:
raise ValueError("SQL is empty")
body = text[:-1].strip() if text.endswith(";") else text
if ";" in body:
raise ValueError("Multiple SQL statements are not allowed")
normalized = _LEADING_COMMENTS.sub("", body).lstrip()
first = normalized.split(None, 1)[0].lower() if normalized else ""
if first not in {"select", "with", "insert"}:
raise ValueError(f"Only SELECT and INSERT SQL are allowed, got {first!r}")
if _FORBIDDEN_SQL.search(normalized):
raise ValueError("Forbidden SQL keyword found")
if first == "with" and not re.search(r"\bselect\b", normalized, re.IGNORECASE):
raise ValueError("WITH statements must be SELECT queries")
return body
def as_dict(self) -> dict[str, str]:
return {
"status": self.status,
"first_keyword": self.first_keyword,
"reason": self.reason,
}
def safe_spark_sql(spark, sql: str):
"""Validate then execute. The ONLY sanctioned path to spark.sql()."""
return spark.sql(assert_select_or_insert(sql))
def _mask(sql: str) -> str:
"""Mask strings, comments, and backtick identifiers before token checks."""
out: list[str] = []
i = 0
state = "normal"
quote = ""
while i < len(sql):
ch = sql[i]
nxt = sql[i + 1] if i + 1 < len(sql) else ""
if state == "normal":
if ch in {"'", '"', "`"}:
state = "quote"
quote = ch
out.append(" ")
i += 1
elif ch == "-" and nxt == "-":
state = "line_comment"
out.extend(" ")
i += 2
elif ch == "/" and nxt == "*":
state = "block_comment"
out.extend(" ")
i += 2
else:
out.append(ch)
i += 1
elif state == "quote":
out.append(" ")
if ch == quote:
if nxt == quote and quote in {"'", '"'}:
out.append(" ")
i += 2
else:
state = "normal"
i += 1
elif ch == "\\" and nxt:
out.append(" ")
i += 2
else:
i += 1
elif state == "line_comment":
out.append("\n" if ch == "\n" else " ")
if ch == "\n":
state = "normal"
i += 1
else:
out.append("\n" if ch == "\n" else " ")
if ch == "*" and nxt == "/":
out.append(" ")
state = "normal"
i += 2
else:
i += 1
return "".join(out)
def _body(masked: str) -> str:
text = masked.strip()
return text[:-1].strip() if text.endswith(";") else text
def first_keyword(sql: str) -> str:
match = re.search(r"\b([A-Za-z_][\w]*)\b", _body(_mask(sql)))
return match.group(1).lower() if match else "unknown"
def validate_sql(sql: str, *, allow_overwrite: bool = False) -> GuardResult:
if not sql or not sql.strip():
return GuardResult("FAIL", "unknown", "SQL is empty")
masked_body = _body(_mask(sql))
first = first_keyword(sql)
if ";" in masked_body:
return GuardResult("FAIL", first, "Multiple SQL statements are not allowed")
if first not in ALLOWED_FIRST:
return GuardResult("FAIL", first, f"Only SELECT, WITH, or INSERT is allowed; got {first!r}")
lowered = masked_body.lower()
found = sorted(word for word in FORBIDDEN if re.search(rf"\b{word}\b", lowered))
if found:
return GuardResult("FAIL", first, "Forbidden keyword found: " + ", ".join(found))
if first == "with" and not re.search(r"\bselect\b", lowered):
return GuardResult("FAIL", first, "WITH must contain SELECT")
if first == "insert":
if not re.search(r"\bselect\b", lowered):
return GuardResult("FAIL", first, "INSERT must be based on SELECT")
if re.search(r"\binsert\s+overwrite\b", lowered) and not allow_overwrite:
return GuardResult("FAIL", first, "INSERT OVERWRITE requires --allow-overwrite")
return GuardResult("PASS", first, "")
def assert_allowed_sql(sql: str, *, allow_overwrite: bool = False) -> str:
result = validate_sql(sql, allow_overwrite=allow_overwrite)
if result.status != "PASS":
raise ValueError(result.reason)
return sql.strip().rstrip(";")
def safe_spark_sql(spark, sql: str, *, allow_overwrite: bool = False):
return spark.sql(assert_allowed_sql(sql, allow_overwrite=allow_overwrite))
@@ -1,64 +1,77 @@
"""Standard test set for sql_guard.assert_select_or_insert.
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
Run from the scripts/ subdirectory of the skill:
cd .claude/skills/pyspark-sql-guardrails/scripts
python3 test_sql_guard.py
from sql_guard import assert_allowed_sql, validate_sql
Or from the project root:
python3 .claude/skills/pyspark-sql-guardrails/scripts/test_sql_guard.py
PASS_CASES = [
("plain SELECT", "SELECT 1"),
("WITH SELECT", "WITH x AS (SELECT 1 AS a) SELECT a FROM x"),
("INSERT INTO SELECT", "INSERT INTO target_table SELECT * FROM source_table"),
("trailing semicolon", "SELECT 1;"),
("leading line comment", "-- harmless\nSELECT 1"),
("leading block comment", "/* harmless */ SELECT 1"),
("danger word in string", "SELECT 'drop table is text' AS note"),
("danger word in double quote", 'SELECT "delete" AS note'),
("danger word in backtick identifier", "SELECT `drop` FROM t"),
("semicolon in string", "SELECT ';' AS semi"),
("semicolon in comment", "-- ;\nSELECT 1"),
]
Any change to assert_select_or_insert must be followed by running this
set: all EXPECT_RAISE cases must raise, all EXPECT_OK cases must return
cleanly. Failures are loud (non-zero exit).
"""
PASS_WITH_OVERWRITE = [
("INSERT OVERWRITE when explicitly allowed", "INSERT OVERWRITE target SELECT * FROM source"),
]
import sys
import os
# Allow running this file directly from the scripts/ subdirectory.
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from sql_guard import assert_select_or_insert
FAIL_CASES = [
("DROP", "DROP TABLE x"),
("UPDATE", "UPDATE t SET a = 1"),
("DELETE", "DELETE FROM x"),
("TRUNCATE", "TRUNCATE TABLE x"),
("ALTER", "ALTER TABLE x ADD COLUMN a int"),
("CREATE", "CREATE TABLE x(id int)"),
("REPLACE", "CREATE OR REPLACE VIEW v AS SELECT 1"),
("MERGE", "MERGE INTO t USING s ON t.id=s.id WHEN MATCHED THEN UPDATE SET a=1"),
("MSCK", "MSCK REPAIR TABLE x"),
("REFRESH", "REFRESH TABLE x"),
("ANALYZE", "ANALYZE TABLE x COMPUTE STATISTICS"),
("CACHE", "CACHE TABLE x"),
("UNCACHE", "UNCACHE TABLE x"),
("VACUUM", "VACUUM x"),
("OPTIMIZE", "OPTIMIZE x"),
("CALL", "CALL system.proc()"),
("GRANT", "GRANT SELECT ON TABLE x TO user"),
("REVOKE", "REVOKE SELECT ON TABLE x FROM user"),
("stacked statements", "SELECT 1; DELETE FROM x"),
("empty", " "),
("WITH without SELECT", "WITH x AS (DELETE FROM t)"),
("INSERT VALUES", "INSERT INTO target VALUES (1)"),
("INSERT OVERWRITE default blocked", "INSERT OVERWRITE target SELECT * FROM source"),
]
def expect_ok(name, sql):
try:
assert_select_or_insert(sql)
print(f" [OK] {name}")
except ValueError as e:
raise AssertionError(f"{name} should have passed but raised: {e}")
def main() -> int:
print("=== EXPECT PASS ===")
for name, sql in PASS_CASES:
result = validate_sql(sql)
assert result.status == "PASS", (name, sql, result)
assert assert_allowed_sql(sql)
print("[OK]", name)
print("\n=== EXPECT PASS WITH OVERWRITE ===")
for name, sql in PASS_WITH_OVERWRITE:
result = validate_sql(sql, allow_overwrite=True)
assert result.status == "PASS", (name, sql, result)
assert assert_allowed_sql(sql, allow_overwrite=True)
print("[OK]", name)
print("\n=== EXPECT FAIL ===")
for name, sql in FAIL_CASES:
result = validate_sql(sql)
assert result.status == "FAIL", (name, sql, result)
print("[OK]", name, "->", result.reason)
print("\nALL TESTS PASSED")
return 0
def expect_raise(name, sql):
try:
assert_select_or_insert(sql)
raise AssertionError(f"{name} should have raised but passed")
except ValueError:
print(f" [OK] {name}: raised as expected")
print("=== EXPECT_OK ===")
expect_ok("plain SELECT", "SELECT 1")
expect_ok("WITH ... SELECT", "WITH t AS (SELECT 1 AS x) SELECT x FROM t")
expect_ok("trailing semicolon", "SELECT 1 FROM dual;")
expect_ok("leading -- comment", "-- comment\nSELECT 1")
expect_ok("leading /* comment", "/* hi */ SELECT 1")
expect_ok("INSERT ... SELECT", "INSERT INTO t SELECT 1")
print()
print("=== EXPECT_RAISE ===")
expect_raise("DROP", "DROP TABLE loan_order")
expect_raise("UPDATE", "UPDATE loan_order SET status = 1")
expect_raise("DELETE", "DELETE FROM loan_order")
expect_raise("TRUNCATE", "TRUNCATE TABLE loan_order")
expect_raise("ALTER", "ALTER TABLE loan_order ADD COLUMN x INT")
expect_raise("MERGE", "MERGE INTO t USING s ON t.id = s.id")
expect_raise("MSCK", "MSCK REPAIR TABLE loan_order")
expect_raise("REFRESH", "REFRESH TABLE loan_order")
expect_raise("VACUUM", "VACUUM TABLE loan_order")
expect_raise("stacked ;", "SELECT 1; DROP TABLE loan_order")
expect_raise("empty", " ")
expect_raise("comment-hidden", "-- ok\nDROP TABLE loan_order")
print()
print("ALL TESTS PASSED")
if __name__ == "__main__":
raise SystemExit(main())
+29 -46
View File
@@ -1,60 +1,43 @@
#!/usr/bin/env python3
"""One-line SQL guard for the pyspark-sql-guardrails skill.
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
Usage:
# Pipe SQL via stdin (preferred for multi-line SQL)
echo "SELECT ..." | python3 validate_sql.py
cat query.sql | python3 validate_sql.py
# Or pass SQL as the first argument (single-line, quoting-friendly)
python3 validate_sql.py "SELECT 1"
Exit code:
0 -> PASS (SQL is allowed)
1 -> FAIL (SQL violates the allowlist; reason printed to stdout)
This script is the ONLY sanctioned way to validate a generated SQL
string before showing it to the user. It is intentionally short so it
can be invoked in a single Bash call from the assistant.
"""
from __future__ import annotations
import argparse
import json
import sys
import os
from pathlib import Path
# Allow running from anywhere; resolve bundled sql_guard.py.
# The skill layout is:
# .claude/skills/pyspark-sql-guardrails/
# ├── SKILL.md
# └── scripts/
# ├── sql_guard.py (this script's sibling)
# ├── validate_sql.py
# └── test_sql_guard.py
SCRIPTS_DIR = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, SCRIPTS_DIR)
from sql_guard import assert_select_or_insert # noqa: E402
from sql_guard import validate_sql
def read_sql() -> str:
if len(sys.argv) > 1:
return sys.argv[1]
def read_sql(args: argparse.Namespace) -> str:
if args.file:
return Path(args.file).read_text(encoding="utf-8")
if args.sql:
return " ".join(args.sql)
if not sys.stdin.isatty():
return sys.stdin.read()
print(__doc__, file=sys.stderr)
sys.exit(2)
raise SystemExit("Provide SQL as arguments, --file, or stdin")
def main() -> int:
sql = read_sql()
try:
body = assert_select_or_insert(sql)
except ValueError as e:
print(f"FAIL - {e}")
return 1
first = body.split(None, 1)[0].lower()
print(f"PASS - first keyword: {first}, body length: {len(body)}")
return 0
parser = argparse.ArgumentParser(description="Validate Spark SQL against a safe allowlist.")
parser.add_argument("sql", nargs="*", help="SQL string; omitted when using --file or stdin")
parser.add_argument("--file", help="Read SQL from a UTF-8 file")
parser.add_argument("--json", action="store_true", help="Emit machine-readable JSON")
parser.add_argument("--allow-overwrite", action="store_true", help="Allow INSERT OVERWRITE ... SELECT")
args = parser.parse_args()
result = validate_sql(read_sql(args), allow_overwrite=args.allow_overwrite)
if args.json:
print(json.dumps(result.as_dict(), ensure_ascii=False))
elif result.status == "PASS":
print(f"PASS - first_keyword={result.first_keyword}")
else:
print(f"FAIL - first_keyword={result.first_keyword} reason={result.reason}")
return 0 if result.status == "PASS" else 1
if __name__ == "__main__":
sys.exit(main())
raise SystemExit(main())
+155 -88
View File
@@ -1,121 +1,188 @@
---
name: pyspark-sql-pipeline
description: Use when orchestrating the full PySpark SQL pipeline from a business requirement to a reviewed, ready-to-execute SQL string. Triggers include "完整跑一遍取数流程", "生成 SQL 并 review", "end-to-end 取数", "full pipeline", "从头到尾", "数据需求到 SQL". Coordinates requirements-analysismetadata-validatorlogic-plannersql-context-builderpyspark-sql-guardrailssql-review as a single deterministic flow with explicit user-confirmation gates between stages. Does NOT execute SQL — it drives the pipeline to a reviewed SQL string.
description: 当用户希望从业务需求端到端生成一条经过校验和评审的 Spark SQL / PySpark SQL 作业时使用。它编排 requirements-analysismetadata-validatorlogic-plannersql-context-builder、SQL 生成、pyspark-sql-guardrailssql-review,并以正式交付格式输出 DRD、元数据校验、逻辑计划、SQL 上下文、最终 SQL、review 和交付清单。
---
# PySpark SQL Pipeline
# PySpark SQL Pipeline(端到端 SQL 流水线)
## Overview
## 目标
End-to-end orchestrator for turning a business data requirement into a reviewed SQL string. Each stage has a fixed input/output contract; between any two stages there is a **user-confirmation gate** when the upstream status is not green.
把“用户一句业务需求”推进到“可交付、可审查、可追溯的 SQL 和配套审查结果”。它不是单个 SQL 生成器,而是一条带关口的工程流程。
Core principle: **never skip a stage, never let a stage quietly proceed past its own `NEED_USER_CONFIRMATION`.** The pipeline is a sequence of hand-offs, not a single agent's internal monologue.
本流水线默认不执行 SQL、不提交 Spark 作业。只有用户明确要求执行,并且 SQL 已通过 guardrails 和 review 后,才允许进入 MCP 工具层。
## When to Use
## 阶段顺序
Use when:
- The user gives a data requirement and wants SQL produced end-to-end (e.g. "统计近 30 天各城市贷款总额及客户数")
- The user asks to "run the full pipeline" or "从需求到 SQL 一条龙"
- A non-trivial SQL change touches multiple skills' contracts
Do NOT use when:
- The user only needs one stage (e.g. "review this SQL" → use `sql-review` directly)
- The user explicitly says "skip validation" or "just give me SQL fast" — surface the risk and ask once; if they confirm, you may collapse stages but you MUST still apply `pyspark-sql-guardrails` and `sql-review`
## The Pipeline
```dot
digraph pipeline {
"1. requirements-analysis" [shape=box];
"2. metadata-validator" [shape=box];
"3. logic-planner" [shape=box];
"4. sql-context-builder" [shape=box];
"5. SQL generation" [shape=box];
"6. pyspark-sql-guardrails" [shape=box];
"7. sql-review" [shape=box];
"1. requirements-analysis" -> "2. metadata-validator" [label="status=READY_FOR_VALIDATION"];
"2. metadata-validator" -> "3. logic-planner" [label="status=VALIDATED"];
"3. logic-planner" -> "4. sql-context-builder" [label="status=PLANNED"];
"4. sql-context-builder" -> "5. SQL generation" [label="status=READY"];
"5. SQL generation" -> "6. pyspark-sql-guardrails" [label="SQL string"];
"6. pyspark-sql-guardrails" -> "7. sql-review" [label="guardrail pass"];
}
```text
1. requirements-analysis
2. metadata-validator
3. logic-planner
4. sql-context-builder
5. SQL generation
6. pyspark-sql-guardrails
7. sql-review
8. optional MCP execution handoff
```
Every arrow is a **contract boundary**. Do not let a stage start before its predecessor has produced the expected status.
## 每阶段放行条件
## Stage Contracts
```yaml
requirements-analysis: requirements_output.status == READY_FOR_METADATA
metadata-validator: validation_result.status == VALIDATED
logic-planner: logic_plan.status == PLANNED
sql-context-builder: sql_context.status == CONTEXT_READY
pyspark-sql-guardrails: sql_guard.status == PASS
sql-review: sql_review.verdict == PASS 或用户明确接受非 HIGH 风险
mcp-handoff: user explicitly requested execution
```
| # | Stage | Skill | Input | Output Status (green) | Output Status (stop) |
|---|---|---|---|---|---|
| 1 | Requirements | `requirements-analysis` | user message | `READY_FOR_VALIDATION` | `NEED_USER_CONFIRMATION` |
| 2 | Metadata | `metadata-validator` | requirements_output + workspace | `VALIDATED` | `NEED_USER_CONFIRMATION` |
| 3 | Logic | `logic-planner` | validation_result | `PLANNED` | `NEED_USER_CONFIRMATION` |
| 4 | Context | `sql-context-builder` | logic_plan + field_mapping | `READY` | `NEED_USER_CONFIRMATION` |
| 5 | Generate | (no skill — see "SQL Generation" below) | sql_context | SQL string | n/a |
| 6 | Guardrail | `pyspark-sql-guardrails` | SQL string | `pass` | `fail` (raise) |
| 7 | Review | `sql-review` | SQL string + sql_context | `verdict: PASS` | `verdict: FAIL` |
任何阶段输出 `NEED_USER_CONFIRMATION``FAIL`,流水线必须暂停。
## Gate Rules (READ THESE CAREFULLY)
## SQL 生成规则
A stage that returns a stop status **must not** be followed by the next stage. Instead:
SQL 生成没有单独 skill。生成 SQL 时只能翻译以下内容:
1. Surface the `pending_questions` from that stage to the user.
2. Stop the pipeline.
3. After the user answers, re-run from the same stage (not from the beginning) using the user's answers as additional input.
4. Once the stage produces its green status, advance to the next stage.
- `requirements_output`
- `validation_result`
- `logic_plan.steps`
- `sql_context.aliases`
- `sql_context.fields`
- `sql_context.joins`
- `sql_context.metrics`
- `sql_context.filters`
- `sql_context.time_windows`
- `sql_context.outputs`
- `sql_context.write_intent`
Stages are idempotent. Re-running a stage with the same inputs must produce the same output; with new inputs, it must produce a new output that supersedes the old one. Carry forward all prior `field_mapping`, `validation_result`, `logic_plan`, and `sql_context` as the durable artifacts of the pipeline.
不能临时新增表、字段、alias、join key 或指标公式。每个 `alias.column` 必须能在 `sql_context.fields` 中找到。
## SQL Generation (Stage 5)
## SQL 时间口径硬规则
There is no dedicated skill for generating SQL. The pipeline performs this step directly, with these constraints:
对“近 N 天按 T-1 完整日”,Spark/Hive SQL 必须使用左闭右开窗口:
- Use the `sql_context` as the **only** source of field names, aliases, and join keys.
- Compose the SQL from the `logic_plan.steps` in order: `source``filter``join``transform``dedupe``window``aggregate``output`.
- Every `<alias>.<column>` in the output must appear in `sql_context.fields`.
- The generated SQL must be runnable through `pyspark-sql-guardrails.safe_spark_sql` without further edits.
```sql
event_time >= date_sub(current_date(), N)
AND event_time < current_date()
```
## User-Confirmation Messages
禁止用下面写法表示 timestamp 完整日窗口:
Each gate should produce a single, structured message in this shape:
```sql
event_time BETWEEN date_add(current_date(), -N) AND date_add(current_date(), -1)
```
如果用户指定自然日、滚动到当前时刻、特定日期或特定时区,必须在 DRD 和 SQL 注释里写明。
## 转化率 SQL 硬规则
漏斗/转化率 SQL 必须保证事件顺序。对“注册后完成付费”,分子应判断存在满足条件的后续付费:
```sql
COUNT(DISTINCT CASE
WHEN ur.user_id IS NOT NULL
AND qp.user_id IS NOT NULL
THEN fl.user_id
END) AS converted_user_cnt
```
其中 `qualified_payment` 这类 CTE 必须在 join 到注册事件后筛选:
```sql
po.pay_status = 'SUCCESS'
AND po.pay_time >= ur.register_time
```
不要先对 `pay_order` 按用户取全局最早支付再和 `register_time` 比较;这会漏掉“历史早付费但注册后也完成付费”的用户。
## 推荐 CTE 结构
对新用户转化率这类需求,优先使用清晰 CTE:
```text
first_login 分母人群:窗口内首次登录用户
registered 注册事件:按用户保留注册时间
qualified_payment 合格付费:注册后成功付费
final_agg 按维度聚合输出
```
## 正式交付格式
流水线完成时,最终回答必须包含以下章节,不能只给 SQL:
```markdown
## Pipeline paused at: <stage name>
**Status:** NEED_USER_CONFIRMATION
**Open questions:**
1. <question 1>
2. <question 2>
**Pending options:** A / B / C / other
**To resume:** answer the questions above, then say "继续" (continue).
## 需求说明
## 元数据校验
## 逻辑计划
## SQL 上下文
## 最终 SQL
## SQL Review
## 交付清单
```
Do not present a stage's output as final when its status is a stop status. Do not bury the `pending_questions` inside a longer narrative.
每章要求:
## Failure Handling
- `需求说明`:业务目标、统计口径、涉及表、表关系、字段清单。
- `元数据校验`:几张表、多少字段、join key、时间字段是否通过。
- `逻辑计划`:用表格列出 CTE/步骤、说明和关键条件。
- `SQL 上下文`:alias、字段来源、join、指标、时间窗口。
- `最终 SQL`:只在 guardrails PASS 且 review 非 FAIL 后展示。
- `SQL Review`:verdict、最高风险等级、风险条数、关键建议。
- `交付清单`SQL 文件、PySpark 脚本、是否已执行、如何运行。
- **Guardrail failure (`pyspark-sql-guardrails`)** — do not retry the same SQL. Return to Stage 5 and regenerate, this time respecting the guardrail.
- **Review verdict FAIL with HIGH findings** — return to Stage 5 (or earlier if a finding is about context, in which case return to Stage 4). Do not "patch" the SQL to silence a finding; address the root cause.
- **Review verdict FAIL with only MEDIUM/LOW** — surface the findings to the user. They decide whether to ship or fix.
- **Inconsistency between stages** (e.g. SQL references a column not in `sql_context`) — treat as a Stage 5 bug; the SQL was generated without honoring the context. Regenerate, do not amend.
如果生成 PySpark 脚本但当前容器没有 Java/Spark,要明确:
## Forbidden Behaviors
```text
已生成脚本,但当前 opencode 容器不是 Spark 执行环境,未本地执行。可在 Spark 集群中使用 spark-submit 运行。
```
| Rationalization | Reality |
|---|---|
| "User asked for SQL fast, skip the gates" | Gates exist because skipping them produces wrong numbers. Ask the user once; if they confirm, you may collapse only adjacent non-validation stages. Never skip guardrail or review. |
| "Just one stage failed, no need to ask" | Every stop status is a question to the user. Always ask. |
| "I'll inline the validation into SQL generation" | Stages are separate so they can be re-run independently. Inlining couples them. |
| "Review found MEDIUM, ship it" | Surface the findings. The user — not the agent — decides. |
| "Field mapping is obvious, skip the validator" | Field mapping is the validator's job. Bypassing it is the #1 cause of hallucinated column names. |
| "Same status, just continue" | Status names are fixed strings. Re-check the actual status text, do not pattern-match on memory. |
## 关口消息
## Completion Criteria
当阶段暂停时,必须用这个格式:
The pipeline is complete only when:
- All 7 stages have run in order
- `sql-review.verdict` is `PASS` (or the user has explicitly accepted remaining MEDIUM/LOW findings)
- The final SQL string, the `sql_context`, the `logic_plan`, the `validation_result`, and the `requirements_output` are all available to the user in the same response
```markdown
## Pipeline paused at: <stage>
**Status:** NEED_USER_CONFIRMATION
**Open questions:**
1. <question>
**To resume:** answer the questions above, then say "继续".
```
If the user later changes the requirement, restart at Stage 1 — do not patch downstream artifacts.
## 输出格式
```yaml
pipeline_result:
status: COMPLETE | PAUSED | FAILED
current_stage: requirements-analysis | metadata-validator | logic-planner | sql-context-builder | sql-generation | pyspark-sql-guardrails | sql-review | mcp-handoff
requirements_output: {}
validation_result: {}
logic_plan: {}
sql_context: {}
sql: "只在 guardrails PASS 后展示"
pyspark_code: "只在 SQL 通过校验和评审后生成"
sql_guard: {}
sql_review: {}
mcp_plan:
should_execute: false
allowed_tools: []
pending_questions: []
blocking_findings: []
```
## 失败处理
- 需求变了:回到 requirements-analysis。
- 缺表缺字段:回到 metadata-validator。
- 指标公式或 join 粒度错:回到 logic-planner。
- SQL 引用了 context 没有的字段:回到 sql-context-builder 或更上游。
- guardrails 失败:重新生成 SQL,不改弱校验器。
- review 有 HIGH:阻断执行,修根因。
- MCP 准备失败:只修执行参数或脚本落盘问题,不反向篡改已评审 SQL。
## 禁止行为
- 不允许跳过 metadata-validator。
- 不允许 guardrails PASS 后跳过 review。
- 不允许 review 失败后只做字符串补丁而不检查上游产物。
- 不允许在用户未确认时继续下一阶段。
- 不允许把 `prepare_submit_job` 当成真正执行成功。
- 不允许在未获用户明确确认时调用 `confirm_submit_job`
+118 -233
View File
@@ -1,273 +1,158 @@
---
name: requirements-analysis
description: 数据需求解读与拆解 skill。当用户提出"统计/计算/分析/取数/拉个数/做个报表/口径/指标/转化率/留存/活跃/DAU/GMV/漏斗/同环比"等数据类需求,或提供数据分析数据开发需求文档,或贴出表 schema 询问怎么写 SQL/Hive/Spark/取数逻辑时使用。本 skill 不直接写 SQL,而是先把需求拆解成可执行的 DRD(数据需求规格说明书):识别核心实体、解析表结构、推导字段、识别歧义口径、构建 join 关系,并主动向用户索要缺失信息直到口径完全明确。无论用户是否提供 schema、是否催促"直接 SQL",都先走完澄清流程再交付
description: 当用户提出统计、取数、报表、指标转化率、漏斗、留存活跃、GMV、同环比、SQL/Hive/Spark/PySpark 数据开发需求时使用。先把模糊业务问题澄清成可交付的数据需求规格说明书 DRD,不直接 SQL,不调用 MCP,不提交作业
---
# 数据需求解读 (Requirements Analysis)
# Requirements Analysis(需求澄清)
## 你的角色
## 目标
你是数据开发工程师面前的"需求接口人"。用户(通常是产品、运营、数据分析师,甚至业务方)抛过来一句模糊的统计需求,你的任务是:把它翻译成另一位数据开发同学拿到就能直接写 SQL 的规格说明书(DRD)
把一句可能含糊的数据需求整理成正式 DRD。重点不是快点写 SQL,而是先把统计对象、时间窗口、指标分子分母、维度、表字段、join 和输出要求固定下来,避免“SQL 看起来对但数字口径错”
这意味着你的产出**不是 SQL**,而是一份口径明确、字段对齐、join 关系清晰的需求文档
本 skill 的输出是 `requirements_output`,下游只能基于它推进,不得重新解释业务口径
## 为什么先澄清再动手
## 工作顺序
数据需求最大的坑不在 SQL 写错,而在口径理解错。"近 30 天新用户转化率"这一句话里,"近 30 天"、"新用户"、"转化"三个词每一个都至少有 3 种合理解释。如果不先澄清就动手,最后跑出来的数字可能和业务方期望差几个数量级,返工成本远高于多问几轮
1. 复述业务目标:这个统计用于评估什么,最终交付什么
2. 定义输出粒度:结果一行代表什么维度组合。
3. 明确统计对象:用户、订单、交易、设备、账户等。
4. 明确时间口径:自然日、T-1 完整日、滚动窗口、时区、事件时间字段。
5. 明确指标口径:分子、分母、去重键、过滤条件、空值策略。
6. 明确维度口径:渠道、城市、产品、日期等分组字段来自哪张表。
7. 明确数据来源:候选表、候选字段、字段含义、状态枚举、金额单位。
8. 明确关联关系:主表、维表、join key、join 类型、1:N 风险。
9. 明确结果要求:输出列、排序、是否落表、是否允许 overwrite。
10. 如果任一关键口径不明确,停止并提问;如果已明确,输出 DRD 和 `requirements_output`
所以本 skill 的核心动作是:**发现歧义 → 主动提问 → 等待确认 → 再推进**。宁可多问一轮,不要默认假设。
## 必问歧义
## 输入分流
- “近 30 天”是自然日、T-1 完整日,还是滚动到当前时刻。
- “新用户”是首次注册、首次登录、首次下单、首次激活还是首次付费。
- “成功”对应哪些状态码,退款、撤销、部分成功是否计入。
- “转化率”的分子、分母、事件顺序、去重键和转化窗口是什么。
- 分组维度来自事实表还是维表;维表缺失时是否保留 NULL。
- 多表 join 后是否会产生重复行,是否要先去重或聚合。
- 输出是临时查询还是要写入目标表。
接到需求后,先判断属于哪种情况:
## 时间口径硬规则
### 情况 A:只有业务需求,没有表结构
例如:"统计近 30 天新用户转化率"
主动索要:
- 涉及哪些表(让用户列出表名即可)
- 这些表的 schema(字段名 + 类型)
- 关键字段的业务含义(特别是状态、时间、金额类)
- 指标口径定义("新用户"、"转化"分别指什么)
### 情况 B:需求 + 单表/少量表 schema
理解业务逻辑 → 推导所需字段 → 检查是否缺关键字段 → 向用户确认口径。
### 情况 C:需求 + 多表 schema
额外要做:
- 区分主表 / 维表 / 事实表
- 推导 join 关系(on 哪些字段、内连接还是左连接)
- 检查 join 字段在两边是否都存在、类型是否一致
## 核心动作(每轮对话都要走)
### 1. 需求拆解
按这 7 个维度把用户的话拆开,**任何一个不明确就要追问**:
对“近 N 天按 T-1 完整日”,统一解释为左闭右开窗口:
```text
业务目标:为什么要看这个数?
统计对象:是用户、订单、商品、还是别的?
统计范围:哪些数据进入统计?哪些被过滤?
统计时间:从什么时候到什么时候?按自然日还是滚动?
统计维度:按什么分组?(渠道 / 城市 / 产品线 / 时间粒度)
统计指标:算 count? sum? 比率?分子分母分别是什么?
输出形式:一个数?一张表?带哪些列?
[start, end) = [current_date() - N days, current_date())
```
### 2. 识别核心实体
Spark/Hive SQL 应表达为:
从需求里抽出涉及的业务实体(用户、订单、商品、商户、设备、渠道、活动、贷款、客户、账户、交易……),并推测每个实体对应哪张表。
```sql
event_time >= date_sub(current_date(), N)
AND event_time < current_date()
```
如果用户没给表名,**列出你的猜测**让用户确认或补充。
不要使用下面写法表示 timestamp 完整日窗口:
### 3. Schema 解析(拿到表结构后)
```sql
event_time BETWEEN date_add(current_date(), -N) AND date_add(current_date(), -1)
```
对每张表,要分析清楚:
因为它很容易只覆盖到结束日期的 00:00:00,漏掉 T-1 白天的数据。
| 维度 | 含义 |
|------|------|
| 表用途 | 这张表记录什么业务事件/状态 |
| 主键 | 唯一标识一行的字段 |
| 业务主键 | 业务上的唯一键(可能和主键不同) |
| 时间字段 | create_time / update_time / 业务时间,用哪个? |
| 状态字段 | 订单状态、用户状态等枚举字段,可取值是什么 |
| 维度字段 | 可用于分组的字段 |
| 金额字段 | 单位是元还是分?是否含税? |
## 转化口径硬规则
### 4. 推导所需字段
转化类需求必须写清:
对照需求拆解,列出每张表要用到的字段。**显式标注哪些字段你假设存在但还没确认**,让用户打勾或补充
- 分母人群,例如“窗口内首次登录的新用户”
- 分子人群,例如“该新用户注册后存在成功付费”。
- 事件顺序,例如 `pay_time >= register_time`
- 后置事件是否也需要在统计窗口内。
- 去重键,例如 `user_id`
### 5. 识别缺失字段
如果用户说“注册后付费”,必须把“后”写进 DRD,不允许只写“注册且付费”。
常见缺失:用户 ID、订单 ID、时间字段、状态字段、金额字段、渠道字段、产品字段。如果发现关键字段缺失,**直接指出**,不要绕开它继续推进。
## 正式 DRD 输出
### 6. 识别需求歧义(重要!)
这一步是本 skill 的核心价值。**对每一个模糊词,主动列出可能的解释让用户选择**。常见歧义点:
**时间口径**
- "近 30 天" → ① 自然日 [今天-30, 今天);② T-30 不含今天 [昨天-30, 昨天];③ 最近 30 个完整日;④ 滚动 30 天
- "本月" → 自然月还是结算月?
- "凌晨数据归哪天" → 0 点切分还是 4 点切分?
**用户口径**
- "新用户" → 注册用户 / 首次下单用户 / 首次登录用户 / 首次激活用户
- "活跃用户" → 登录过 / 操作过 / 下单过
- "去重" → 按 user_id 还是按设备 ID 去重?
**状态/转化口径**
- "转化" → 下单 / 支付成功 / 放款 / 激活 / 完成首单
- "成功订单" → 状态码具体是哪些?是否包含部分退款?
- "支付金额" → 应付 / 实付 / 实收(扣手续费后)
**关联口径**
- "用户的订单" → 创建人 / 收货人 / 实际付款人?
- 多表 join → inner / left / 是否要去重防止笛卡尔积?
输出形式:
当关键口径已明确时,必须先输出可读 DRD,再输出 YAML 摘要。
```markdown
## 待确认问题
## 需求说明
1. "新用户"指的是?
- [ ] A. 在统计窗口内首次注册的用户
- [ ] B. 在统计窗口内首次下单的用户
- [ ] C. 其他(请说明)
### 业务目标
...
2. "近 30 天"是?
- [ ] A. [今天-30, 今天) 含今天
- [ ] B. [昨天-30, 昨天] T-1 口径
- [ ] C. 其他
### 统计口径
- 统计对象:...
- 时间范围:近 N 天 T-1 完整日,`event_time >= date_sub(current_date(), N)``event_time < current_date()`
- 维度:...
- 指标:...
### 涉及表
| 表名 | 用途 | 粒度 | 使用字段 |
|---|---|---|---|
### 表关系
| 左表字段 | 右表字段 | Join 类型 | 风险 |
|---|---|---|---|
### 字段清单
| 表名 | 字段 | 用途 | 角色 |
|---|---|---|---|
### 实现逻辑
1. ...
### 待确认问题
无,或列出阻塞问题。
```
### 7. 构建 join 关系
多表场景下,明确画出关联:
```
user_info.user_id = order_info.user_id (left join, 左表为主)
order_info.order_id = pay_info.order_id (inner join)
```
要标明:
- 关联字段及类型是否一致
- join 类型(inner / left / right / full
- 是否会产生一对多导致重复计算
### 8. 确认机制
**严格禁止**
- 自行脑补未说明的业务逻辑
- 默认字段含义(哪怕字段名看上去很标准)
- 默认时间口径、用户口径、转化口径
- 在关键问题没确认前就给出最终 SQL
每轮回复都要包含三块:
-**已确认**:到目前为止双方对齐的内容
-**待确认**:还需要用户回答的问题(编号列出,方便用户对照回答)
- 📋 **当前推导**:基于已知信息你推出的字段清单 / join 关系(标注哪些是假设)
## 完成判定
只有以下 7 项**全部**明确,才算需求分析完成、可以交付 DRD:
- [ ] 业务逻辑明确
- [ ] 指标口径明确(分子分母、聚合方式)
- [ ] 时间范围明确(窗口定义、时区、是否含端点)
- [ ] 维度明确(group by 哪些字段)
- [ ] 表来源明确(每个数据来自哪张表)
- [ ] join 关系明确(关联键、关联类型)
- [ ] 字段明确(每个字段都已被用户确认存在)
任何一项打问号,都继续走澄清流程,不要交付最终 DRD。
## 最终交付:DRD 模板
当 7 项全部确认后,按这个模板输出:
```markdown
# 数据需求规格说明书 (DRD)
## 一、业务目标
(1-2 句话说明这个数据是做什么用的、给谁看、支撑什么决策)
## 二、统计口径
- 统计对象:xxx
- 时间范围:xxx(精确到时区、端点)
- 过滤条件:xxx
- 聚合维度:xxx
- 指标定义:
- 指标 A = 分子 / 分母,分子定义为 xxx,分母定义为 xxx
## 三、涉及表
### 表 1<table_name>
- 用途:xxx
- 粒度:一行代表 xxx
- 使用字段:
- `field_a`: 含义
- `field_b`: 含义
### 表 2<table_name>
(同上)
## 四、表关联关系
```
table_a.id = table_b.a_id (left join)
```
说明:xxx
## 五、最终字段清单
| 表名 | 字段名 | 用途 | 备注 |
|------|--------|------|------|
| table_a | id | 主键 | |
| table_a | created_at | 时间过滤 | UTC+8 |
## 六、实现逻辑(伪代码层面)
1. 从 table_a 取 [time_range] 内的数据,过滤 status = xxx
2. 按 user_id left join table_b
3. group by xxx,聚合 xxx
4. 输出列:xxx
## 七、待开发 SQL 所需信息
全部已确认 ✓
```
## 风格提醒
- 每轮回复都用结构化 markdown,不要长段落散文
- 提问要给选项(A/B/C),降低用户回答成本
- 字段名、表名用反引号 `code` 包起来
- 时间相关问题特别仔细,时区、端点、自然日 vs 滚动这三件事最容易翻车
- 如果用户催"直接给 SQL",温和坚持:"为了避免数字跑出来不对,先把这 N 个口径敲定,几分钟就能确认完,然后直接交付准确的 SQL"
- 始终记得:你的产出物是 DRD,不是 SQL
## 例外情况
如果用户**明确表示**"我已经想清楚口径了,不用再问,按以下定义直接生成 SQL",并且把所有 7 项完成判定都写明了,那么可以跳过澄清直接进入 DRD/SQL 阶段。但即便这样,也要在回复里复述一遍你理解的口径,让用户最后过一眼。
---
## 下游契约(Pipeline Handoff
本 skill 是 PySpark SQL 流水线的**第一站**。澄清完成后,必须把结构化结果以标准化形式交付给 `metadata-validator`,由其继续推进。
**Required next step:** `metadata-validator` — 它会递归扫描工作区元数据文件并校验候选表/字段/关联/时间字段是否真实存在。
在交付 DRD 之前,必须同时输出以下标准化结构(与第七节 DRD 内容一致):
## 结构化输出
```yaml
requirements_output:
business_goal: "<一句话业务目标>"
business_logic: "<一段话业务逻辑,含统计对象/时间窗口/分组/指标分子分母>"
candidate_tables:
- <table_name> # 用户提及或根据业务推断的物理表
candidate_fields:
- name: <business_field_name> # 业务字段名(如 客户号 / 贷款金额)
suggested_table: <table_name> # 建议所在表
suggested_column: <column_name> # 建议字段名(如未确认可省略)
role: <dimension|metric|filter|time|join_key>
status: READY_FOR_METADATA | NEED_USER_CONFIRMATION
business_goal: "一句话业务目标"
grain: "结果一行代表什么"
time_window:
field_hint: <table.column> # 用户提到的时间字段
window: "<窗口描述,如 近 30 天 / 2026-05-01 ~ 2026-05-31>"
join_hints: # 用户已说明的关联(如有)
- left_table: <table>
left_field: <column>
right_table: <table>
right_field: <column>
join_type: <inner|left|right|full>
status: <READY_FOR_VALIDATION | NEED_USER_CONFIRMATION>
description: "时间窗口"
timezone: "时区"
event_time_hint: "候选时间字段"
lower_bound_sql: "date_sub(current_date(), N)"
upper_bound_sql: "current_date()"
boundary: "left_closed_right_open"
metrics:
- name: "指标名"
formula: "业务公式"
numerator: "分子"
denominator: "分母"
dedupe_key: "去重键"
order_constraint: "例如 pay_time >= register_time"
null_policy: "空值处理"
dimensions:
- "分组维度"
filters:
- "过滤条件"
candidate_tables:
- business_role: "事实表/维表/码表/结果表"
table_hint: "候选物理表"
candidate_fields:
- business_name: "业务字段名"
role: dimension | metric | filter | time | join_key | output
table_hint: "候选表"
column_hint: "候选字段"
required: true
join_hints:
- left: "表A.字段"
right: "表B.字段"
type_hint: left | inner | right | full | unknown
cardinality_hint: "1:1 | 1:N | N:1 | N:N | unknown"
output_columns:
- "输出列"
write_intent:
mode: query_only | insert_into | insert_overwrite | unknown
target_table: "可为空"
overwrite_confirmed: false
pending_questions: []
```
**契约约束:**
## 停止条件
- `status: READY_FOR_VALIDATION` 只有在第 7 项完成判定全部打勾时才允许输出
- `status: NEED_USER_CONFIRMATION` 时,必须同时在回复里保留第七节的 DRD「待确认问题」清单,`metadata-validator` **不得**在澄清完成前介入。
- `candidate_tables` / `candidate_fields``metadata-validator` 的**唯一**输入;不要让 `metadata-validator` 自行反推业务字段。
- 本 skill 不直接进入 SQL 生成。任何“直接给 SQL”的要求都必须先走完澄清 → DRD → `metadata-validator``logic-planner``sql-context-builder` 这条链路。
只要 `pending_questions` 非空,就输出 `NEED_USER_CONFIRMATION`,不要进入元数据校验,更不要写 SQL
+10 -16
View File
@@ -2,25 +2,19 @@
"skill_name": "requirements-analysis",
"evals": [
{
"id": 1,
"name": "vague-no-schema",
"prompt": "帮我统计下近30天新用户转化率",
"expected_output": "应主动索要相关表/schema/字段说明,并对'近30天'、'新用户'、'转化'这三个词分别提出多选式澄清问题。不应直接生成 SQL 或假设口径。",
"files": []
"id": "ambiguous-conversion-rate",
"prompt": "帮我看下最近一个月新用户转化率",
"expected_behavior": "必须追问最近一个月的时间边界、新用户定义、转化动作、分子分母、去重键和候选表字段;不能直接写 SQL。"
},
{
"id": 2,
"name": "schema-with-ambiguity",
"prompt": "统计上周每个渠道的支付成功订单数和支付金额。表结构如下:\n\norder_info(\n order_id bigint,\n user_id bigint,\n channel_id int,\n create_time timestamp,\n pay_time timestamp,\n status int, -- 1待支付 2已支付 3已发货 4已完成 5已退款 6部分退款\n amount decimal(18,2), -- 应付金额\n paid_amount decimal(18,2) -- 实付金额\n)\n\nchannel_dim(\n channel_id int,\n channel_name varchar\n)",
"expected_output": "应识别出'支付成功'状态码歧义(2/3/4/6 是否都算)、'支付金额'是 amount 还是 paid_amount、'上周'是自然周还是滚动 7 天、退款订单是否计入。应提议 left join channel_dim。不应直接生成 SQL。",
"files": []
"id": "orders-with-schema",
"prompt": "统计上周各渠道支付成功订单数和金额,并给出 order_info/channel_dim schema",
"expected_behavior": "必须确认上周口径、支付成功状态、金额字段、退款是否计入、join 类型和输出粒度。"
},
{
"id": 3,
"name": "multi-table-loan",
"prompt": "我要算下贷款产品的首贷转化漏斗,从注册到首次申请到首次放款的转化率,按月看趋势。表:\n\nuser_register(user_id bigint, register_time timestamp, channel varchar)\nloan_apply(apply_id bigint, user_id bigint, apply_time timestamp, product_id int, apply_amount decimal)\nloan_disburse(disburse_id bigint, apply_id bigint, disburse_time timestamp, disburse_amount decimal, status int)\nproduct_dim(product_id int, product_name varchar, product_type varchar)",
"expected_output": "应识别多表 join 关系(user → apply 按 user_id, apply → disburse 按 apply_id),追问首贷定义(按用户首次还是按产品首次)、放款 status 含义、'按月看'是注册月还是事件发生月、是否限定产品类型、漏斗分母是否包含没注册成功的用户。",
"files": []
"id": "loan-funnel",
"prompt": "做注册到申请到放款的贷款漏斗,按月看",
"expected_behavior": "必须确认月份归属、首贷定义、放款成功状态、漏斗分母、用户去重和 user/apply/disburse 关联关系。"
}
]
}
}
+557
View File
@@ -0,0 +1,557 @@
---
name: spark-executor-mcp-operate
description: Operation guide for the spark-executor-mcp service — 16 MCP tools that submit, monitor, fetch logs from, and kill PySpark jobs on YARN, plus connection-profile and script-file management. USE THIS SKILL whenever the user wants to do anything with the spark-executor-mcp MCP service: submit a Spark/PySpark job to YARN, check job status or result, fetch container logs, kill a running job, save/list/get/delete a YARN connection, or write/read/update a PySpark script file. Also use it when the user asks generally about Spark, YARN, pyspark, cluster jobs, or distributed compute via this service. Do NOT guess tool names or invent a submit flow — read this skill first to learn the two-step prepare/confirm flow, the dual-ID contract (job_id vs application_id), and the SQL guard / path restrictions. The MCP service does NOT generate PySpark code — the calling LLM writes the code, then uses `write_job_file` to persist it on disk before submitting.
---
# spark-executor-mcp 操作指南
Spark-executor-mcp is an MCP (Model Context Protocol) service that exposes
**16 tools** for managing PySpark jobs on YARN. It runs as a FastAPI app
mounted at `/spark-executor-mcp`; clients (typically LLM agents) call the
tools via the standard MCP `tools/call` flow.
This skill is the canonical reference for an LLM agent using the service.
It does **not** describe implementation details — for that, see the source
in `spark_executor/`.
---
## 0. Mindset before you start
Three things you MUST internalize before calling any tool:
1. **The submit flow is two-step: `prepare_submit_job` then `confirm_submit_job`.**
Prepare is cheap and reversible (just snapshots params + writes JSON to
disk). Confirm is irreversible — it actually invokes `spark-submit` and
a YARN application starts running. Never skip the prepare step, even
when the user is in a hurry.
2. **Every Job has two IDs that are NOT interchangeable everywhere.**
- `job_id` — local, 12-char hex (`a1b2c3d4e5f6`), generated by the
service. Used for: looking up the Job record, all MCP tool calls.
- `application_id` — YARN's own ID (`application_17400000001_0001`).
Used for: YARN UI, `tracking_url`, the YARN REST API.
The service's 4 job-lifecycle tools accept **either** ID (after the
2026-06-29 fix), but other systems (YARN UI, kubectl-style scripts)
only know `application_id`. **Always store both** from every
`confirm_submit_job` response.
3. **The LLM writes the PySpark code. The MCP service writes the file.**
`write_job_file` is a file writer with SQL-safety checks, not a
code generator. You compose the code in your context, then call
`write_job_file(code=...)` to drop it on the container's
filesystem where `spark-submit` can see it.
4. **Three decisions MUST be approved by the human user before you act on
them — never pick defaults silently, never auto-confirm.** See §0.1
below. This is non-negotiable: confirm_submit_job starts a real YARN
job that costs cluster time and may write to production data paths,
and the user is the one who has to live with the result. Get their
explicit "yes, confirm" before pulling the trigger.
### 0.1. Mandatory user confirmations (the three checkpoints)
For every Spark job the user asks you to run, you must stop and ask the
human user to approve these three decisions in order. Do not pick
defaults on their behalf. Do not assume "they'd obviously want prod /
default queue / 2G / 2 cores / 2 executors" — those are YOUR defaults,
not theirs.
**Checkpoint 1 — when the user first asks to run a job:**
ASK the user for:
- Which **connection** to submit to. Show the existing connections
(from `list_connections`) as options; if the right one doesn't exist,
ask for the new connection's full profile (name, master, deploy_mode,
yarn_rm_url, spark_conf, auth). See §5 for the field reference.
- The **app_name** (human-readable label for the YARN app). The
service has no default for this; without it `prepare_submit_job`
returns 400.
Do not skip past this even if the user is in a hurry or the request
seems obvious. "Run a daily job on prod" still needs you to confirm
which "daily job", which "prod", and what app_name to label it with.
**Checkpoint 2 — before `prepare_submit_job`:**
ASK the user for the values of these submit parameters, even if they
have server-side defaults:
- `app_name` (if you didn't get it at Checkpoint 1)
- `queue`
- `executor_memory`
- `executor_cores`
- `num_executors`
- any `extra_args` (jars, py-files, …) needed for this script
Tell the user the server-side defaults you WOULD use if they don't care
(say "I'll use default queue, 2G memory, 2 cores, 2 executors unless
you say otherwise") so they can either approve or override. The
service itself rejects implicit defaults (returns 400 listing the
assumed values) — this is by design, and you should respect it by
being the one to surface the assumed values, not by silently letting
the service 400 and then re-asking.
**Checkpoint 3 — before `confirm_submit_job` (the irreversible one):**
Present a one-screen summary of exactly what will happen, and wait for
the user to say "confirm" / "yes" / "do it" (or any unambiguous
affirmative). The summary must include:
- The connection name and its master / deploy_mode / yarn_rm_url
- The app_name and the script path
- queue, executor_memory, executor_cores, num_executors
- The YARN tracking URL that will be returned (you don't know it yet,
but you can say "you'll get a tracking URL once it starts")
The user's exact words don't need to be "confirm" — any clear
affirmative works ("go", "ship it", "y", "🚀", "do it"). A non-answer
or a "wait" / "hold on" / "let me think" is NOT consent; wait.
After they say yes, you may call `confirm_submit_job`. **Never call
`confirm_submit_job` without that explicit go-ahead.**
---
## 1. Tool index
| Operation ID | HTTP | Group | Purpose |
|---|---|---|---|
| `save_connection` | POST | Connection | Upsert a named YARN connection (master, deploy_mode, spark_conf, …) |
| `list_connections` | POST | Connection | List all saved connections |
| `get_connection` | POST | Connection | Fetch one connection by name |
| `delete_connection` | POST | Connection | Delete a connection by name |
| `write_job_file` | POST | Job file | Write LLM-written PySpark code to disk |
| `read_job_file` | POST | Job file | Read a script file back (capped at 1 MB) |
| `update_job_file` | POST | Job file | Overwrite a script file (capped at 1 MB) |
| `prepare_submit_job` | POST | Submit | Snapshot a job to `PendingSubmission`; **does NOT submit** |
| `confirm_submit_job` | POST | Submit | Actually invoke `spark-submit` for a `pending_id` |
| `update_pending_job` | POST | Submit | Edit a `PENDING` submission (script_path, queue, …) |
| `cancel_pending_job` | POST | Submit | Cancel a `PENDING` submission |
| `list_pending_jobs` | POST | Submit | List every PendingSubmission regardless of status |
| `get_pending_job` | POST | Submit | Fetch one PendingSubmission by id |
| `get_job_status` | POST | Job lifecycle | Query YARN for a job's live state (RUNNING / SUCCEEDED / …) |
| `get_job_result` | POST | Job lifecycle | Terminal-oriented view: final_status, diagnostics, timing |
| `get_job_logs` | POST | Job lifecycle | Aggregated container logs (default last 5000 chars) |
| `kill_job` | POST | Job lifecycle | PUT state=KILLED to YARN REST for the app |
(Total: 16 MCP tools + `/health` GET, which is not exposed as a tool.)
---
## 2. The canonical happy path: submit a PySpark job
```
-- CHECKPOINT 1: ask user for connection + app_name --
1. list_connections / save_connection # pick or create the right profile
2. write_job_file # write the LLM's PySpark to disk
3. read_job_file # (optional) sanity-check what was written
-- CHECKPOINT 2: ask user for queue / memory / cores / num_executors --
4. prepare_submit_job # snapshot + persist; no YARN call yet
-- CHECKPOINT 3: present summary, wait for explicit "yes, confirm" --
5. confirm_submit_job # NOW spark-submit runs (irreversible)
6. get_job_status / logs # poll while running
7. get_job_result # terminal view when done
```
The three `CHECKPOINT` lines above are mandatory pauses. Do NOT
collapse them or skip past them. See §0.1 for the full rules; the
short version is: ask the user before each tool call that affects
cluster state, and never call `confirm_submit_job` without their
explicit go-ahead.
### 2.1 Save a connection (one-time per cluster)
```python
save_connection(
name="prod",
master="yarn",
deploy_mode="cluster",
yarn_rm_url="http://rm.example.com:8088",
spark_conf={"spark.sql.shuffle.partitions": "200"},
# optional: auth_type="kerberos", ssl_verify=False, ...
)
```
**`name`** is the lookup key — keep it short and stable. Re-calling
`save_connection` with the same `name` overwrites; the original `yarn_rm_url`
and `spark_conf` are **snapshot into the PendingSubmission at prepare time**
and survive later edits to the connection (this is intentional — the user
can re-point the connection at a different cluster without retargeting
in-flight submissions).
### 2.2 Write the script to disk
The LLM writes the PySpark code. Then:
```python
script_path = write_job_file(code=PYSPARK_SOURCE)["script_path"]
# Returns {"script_path": "/absolute/path/in/container.py", "bytes": N}
```
The path lives inside the container's `SPARK_EXECUTOR_JOBS_DIR` (default
`./data/jobs/`). For pre-existing scripts on the host, mount the host
directory into the container and skip this step — pass the in-container
path directly to `prepare_submit_job`.
### 2.3 Prepare (snapshot, no YARN call)
```python
prepare_submit_job(
connection="prod",
script_path=script_path,
app_name="daily-aggregation", # REQUIRED, no default
queue="default", # these 4 must be explicit
executor_memory="2G",
executor_cores=2,
num_executors=2,
# extra_args={"jars": "hdfs://...jar"}, # optional
)
# Returns {"pending_id": "p_a1b2c3", "status": "PENDING", "parameters": {...}}
```
**You MUST pass `queue`, `executor_memory`, `executor_cores`, `num_executors`
explicitly every time** — even though defaults exist server-side, the
service rejects implicit defaults with a 400 listing the assumed values.
This forces the agent (or the human reviewer) to see and approve what
will actually be submitted.
The response is the only chance to capture the `pending_id` you need for
`confirm_submit_job`. Store it.
### 2.4 (Optional) Edit before submit
If something needs to change between prepare and confirm:
```python
update_pending_job(
pending_id="p_a1b2c3",
queue="research",
# script_path can also be changed — but the new file MUST exist
# and pass the SQL guard
)
```
Refuses to edit anything that is no longer `PENDING`.
### 2.5 Confirm (the irreversible step)
```python
result = confirm_submit_job(pending_id="p_a1b2c3")
# Returns {"job_id": "a1b2c3d4e5f6",
# "application_id": "application_17400000001_0001",
# "tracking_url": "http://rm:8088/proxy/application_17400000001_0001/"}
```
**STORE BOTH IDS NOW.** Future calls to `get_job_status` / `get_job_logs`
/ `kill_job` accept either, but `application_id` is the one YARN itself
recognizes.
`confirm_submit_job` is **idempotent** on `SUBMITTED` (re-calling returns
the same record) and **retries** on transient `SparkSubmitError` (3
attempts with 5 s delay by default). It transitions to `FAILED` only
after all retries are exhausted.
### 2.6 Monitor / fetch / kill
```python
# While running
state = get_job_status(job_id="a1b2c3d4e5f6").state
logs = get_job_logs(job_id="a1b2c3d4e5f6", tail_chars=10000)
# When done
result = get_job_result(job_id="a1b2c3d4e5f6")
# result.state -> "SUCCEEDED" / "FAILED" / "KILLED" / ...
# result.final_status -> "SUCCEEDED" / "FAILED" / "KILLED" / "UNDEFINED"
# result.diagnostics -> YARN's "why" string (often empty on success)
# result.tracking_url -> the same proxy URL confirm_submit_job returned
# To stop it
kill_job(job_id="a1b2c3d4e5f6")
```
All four accept **either** `job_id` or `application_id`. Pass whichever
you have on hand; the service tries `job_id` first, falls back to
`application_id`.
---
## 3. The PendingSubmission state machine
```
prepare_submit_job
|
v
+-------+ cancel_pending_job
|PENDING+----------> CANCELLED (terminal)
+---+---+
| confirm_submit_job
v
+----------+
| SUBMITTED| (terminal — kill via kill_job, not cancel)
+----------+
(or)
+--------+
| FAILED | <-- confirm_submit_job exhausted retries
+--------+
| confirm_submit_job (resets to PENDING, retries)
v
PENDING
```
- `cancel_pending_job` **refuses** to cancel `SUBMITTED` or `FAILED`
records — those are terminal; use `kill_job` instead.
- `update_pending_job` **refuses** to edit anything that is not
`PENDING`.
- `confirm_submit_job` on a `SUBMITTED` record is a no-op (returns the
same `job_id` / `application_id`). On a `FAILED` record, it resets
to `PENDING` and retries. On a `CANCELLED` record, it raises 400.
---
## 4. The dual-ID contract
| ID | Where it comes from | What it's good for |
|---|---|---|
| `job_id` | service-generated, 12-char hex (`a1b2c3d4e5f6`) | Looking up the Job in this service's store; all 4 lifecycle tools |
| `application_id` | parsed from `spark-submit` stderr (`application_<ts>_<n>`) | YARN UI, YARN REST, `tracking_url`, talking to ops scripts |
**Both IDs are returned by `confirm_submit_job`.** Both are accepted by
`get_job_status` / `get_job_result` / `get_job_logs` / `kill_job`. **You
don't need to remember which to pass — but DO remember to store both
from the confirm response** because:
- If the Job store has been wiped (rare; only on data dir loss), only
`application_id` is still meaningful to YARN — you can't poll a YARN
app you can't identify.
- If you're going to paste a link to a teammate, `tracking_url` (derived
from `application_id`) is the human-friendly form.
If a lifecycle tool returns `KeyError("No Job found for id='X' …")`,
check that `X` came from a successful `confirm_submit_job` response.
The error message intentionally mentions both ID forms and shows an
example of each so you can self-diagnose.
---
## 5. Connection profiles
`Connection` is the cluster profile — it's the "where to submit" record.
Fields:
| Field | Required | Default | Notes |
|---|---|---|---|
| `name` | yes | — | The lookup key |
| `master` | yes | `"yarn"` | `"yarn"` / `"spark://…"` / `"k8s://…"` / `"local[N]"` |
| `deploy_mode` | yes | `"cluster"` | `"cluster"` or `"client"` |
| `yarn_rm_url` | no | env `YARN_RESOURCE_MANAGER_URL` | Required if you'll call `get_job_status` / `get_job_logs` / `kill_job` (the lifecycle tools need a YARN endpoint) |
| `spark_conf` | no | `{}` | Merged into `spark-submit --conf` at submit time |
| `ssl_verify` | no | env `SPARK_EXECUTOR_SSL_VERIFY_DEFAULT` | bool |
| `ssl_ca_bundle` | no | env `SPARK_EXECUTOR_SSL_CA_BUNDLE_DEFAULT` | Path to CA file |
| `auth_type` | no | `"none"` | `"none"` / `"simple"` / `"basic"` / `"kerberos"` |
| `auth_user` / `auth_password` | for `basic` | — | |
| `auth_principal` / `auth_keytab` | for `kerberos` (display only) | — | Real auth uses the system cache |
**Snapshot semantics:** when `prepare_submit_job` runs, it copies
`master` / `deploy_mode` / `spark_conf` / `yarn_rm_url` from the
**current** Connection into the `PendingSubmission`. Editing the
Connection later does **not** retarget an in-flight submission.
**Lifecycle tools use the Job's snapshotted `yarn_rm_url`** (set at
confirm time) — not the Connection's current value. If the RM moves
between confirm and a follow-up `get_job_logs`, you have a problem;
re-pointing the Connection does not help.
---
## 6. Job file workflow (`write_job_file` / `read_job_file` / `update_job_file`)
These are **filesystem I/O tools, not code generators**. The LLM in
context writes the PySpark code; these tools just persist it where
`spark-submit` can find it, with safety guards.
**Typical lifecycle:**
```
LLM composes code --> write_job_file(code=...) # write
LLM reviews code --> read_job_file(script_path=...) # verify
LLM revises code --> update_job_file(script_path=..., # overwrite
content=...)
LLM submits --> prepare_submit_job(script_path=...)
```
**Guards (these are why the tools exist instead of just `cat > file.sh`):**
- **SQL guard** — `write_job_file` and `update_job_file` both reject
code containing forbidden SQL (DROP, DELETE, UPDATE, TRUNCATE, ALTER,
GRANT, REVOKE, …). The check is conservative; if it false-positives,
rewrite to use only SELECT/INSERT or restructure.
- **Path restriction** — `update_job_file` only writes under
`SPARK_EXECUTOR_JOBS_DIR`. This blocks accidentally overwriting
host-mounted configs or the service's own files.
- **Size cap** — 1 MB on both `read_job_file` and `update_job_file`.
Use `read_job_file` for small scripts; for huge ones, mount the host
directory and skip these tools.
**For pre-existing scripts on the host:** mount the host dir into the
container (e.g. `-v /host/scripts:/app/scripts:ro` in `docker run`) and
pass the in-container path directly to `prepare_submit_job` — no need
to round-trip through the file tools.
---
## 7. Error reference
| Status | When | What to do |
|---|---|---|
| 400 | `prepare_submit_job` without explicit defaults; `cancel_pending_job` on terminal state; `update_pending_job` on non-PENDING; `prepare_submit_job` with missing/non-file `script_path`; SQL policy violation | Read the message — it almost always names the fix. Re-submit with corrected args. |
| 404 | `KeyError` from tool layer: unknown `connection`, `pending_id`, `job_id`, or `application_id` | The ID you sent doesn't match anything in the store. For lifecycle tools, the error message explicitly suggests both ID forms. |
| 422 | Pydantic validation: missing field, wrong type | Usually a typo or missing required arg. The response body lists every field error. |
| 500 | spark-submit subprocess failed and was not a `SparkSubmitError` (e.g. binary missing); YARN REST call failed | The tool logs the full traceback to `data/logs/error/...` — check there. |
The service translates exceptions at the FastAPI layer:
- `KeyError` → 404 (with the KeyError's string as `detail`)
- `ValueError` → 400
- Pydantic validation → 422
So when the server returns 404, the body `detail` field is the same
"Unknown job_id: …" / "Unknown connection: …" string the tool layer
raised.
---
## 8. Common pitfalls
| Pitfall | Why it bites | What to do |
|---|---|---|
| Skipping the prepare step | `confirm_submit_job` requires a `pending_id`; calling it without one returns 400 | Always prepare first |
| Passing implicit defaults | Server returns 400 listing what you didn't confirm | Pass `queue`, `executor_memory`, `executor_cores`, `num_executors` every time |
| `script_path` doesn't exist in the container | `prepare_submit_job` returns 400; the agent generated the path in its own context but the container's filesystem is different | Either call `write_job_file` first, or mount a host dir |
| SQL injection in the script | `write_job_file` / `update_job_file` / `prepare_submit_job` all 400 | Only use SELECT/INSERT; never DROP/DELETE/UPDATE |
| `cancel_pending_job` on a running YARN app | Returns 400 — submissions are terminal once submitted | Use `kill_job(job_id=...)` instead, which talks to YARN REST |
| Re-pointing a Connection and expecting in-flight jobs to follow | They don't — the Job snapshots `yarn_rm_url` at confirm time | Re-issue with a new `prepare_submit_job` if you actually need a different RM |
| Calling lifecycle tools with the wrong ID type | Used to fail; **fixed in the 2026-06-29 update** — both IDs are accepted | Just pass whichever you have; the service tries `job_id` first |
| Auto-confirming without the user's explicit "yes" | Triggers an irreversible YARN job the user may not have wanted — wastes cluster time, may write to wrong data paths | Per §0.1, **always** present a one-screen summary and wait for the user to say "confirm" / "yes" / equivalent before calling `confirm_submit_job`. Silence, "hmm", or "let me think" are NOT consent. |
| Picking connection / queue / executor defaults on the user's behalf | The user knows their cluster's cost / SLA / data layout better than you do; your defaults may not match their intent | Per §0.1, ASK at Checkpoint 1 (connection + app_name) and Checkpoint 2 (queue / memory / cores / num_executors). Surface the defaults you'd use so they can approve or override, don't just pick. |
| "Session not found" at the MCP layer | Multi-worker gunicorn + in-process session state | Out of scope for this skill; the service warns on startup if `GUNICORN_WORKERS>1`; set `GUNICORN_WORKERS=1` or front with a sticky LB |
---
## 9. End-to-end example (with the three checkpoints)
User: "Run a word count on `s3a://my-bucket/books/*.txt` and write the
result back to `s3a://my-bucket/wc/`."
```
# ============================================================
# CHECKPOINT 1 — ask the user for connection + app_name FIRST
# ============================================================
#
# Agent: "I see you have these saved connections:
# - prod (yarn RM http://rm.prod.example.com:8088, cluster, kerberos)
# - dev (yarn RM http://rm.dev.example.com:8088, client, none)
# Which one should I submit to? And what app_name should I label
# the YARN app with?"
# User: "prod, call it wordcount-books"
#
# (If neither connection is right, ask for the full profile and call
# save_connection before continuing — see §2.1 for the fields.)
# 1. Compose the PySpark code IN YOUR CONTEXT
pyspark_code = '''
from pyspark.sql import SparkSession
spark = SparkSession.builder.appName("wordcount").getOrCreate()
sc = spark.sparkContext
files = sc.wholeTextFiles("s3a://my-bucket/books/")
words = files.flatMap(lambda f: f[1].lower().split()) \\
.map(lambda w: (w, 1)) \\
.reduceByKey(lambda a, b: a + b)
df = words.toDF(["word", "count"])
df.write.mode("overwrite").parquet("s3a://my-bucket/wc/")
'''
# 2. Persist to disk
script_path = write_job_file(code=pyspark_code)["script_path"]
# 3. Sanity-check (optional but recommended)
contents = read_job_file(script_path=script_path)["content"]
# ... review in context ...
# ============================================================
# CHECKPOINT 2 — ask the user for submit parameters
# ============================================================
#
# Agent: "I'll use these unless you say otherwise:
# queue=default, executor_memory=2G, executor_cores=2,
# num_executors=2. Any overrides (different queue, more memory,
# jars/py-files to add)?"
# User: "use research queue, 4G memory, 2 executors"
#
# (If the user says "defaults are fine", that's still an explicit
# approval — you may then call prepare_submit_job with those values.)
# 4. Prepare (snapshot — no YARN call yet)
pending = prepare_submit_job(
connection="prod",
script_path=script_path,
app_name="wordcount-books",
queue="research",
executor_memory="4G",
executor_cores=2,
num_executors=2,
)
pending_id = pending["pending_id"]
# ============================================================
# CHECKPOINT 3 — present summary, wait for explicit "confirm"
# ============================================================
#
# Agent: "About to run on YARN:
# • connection: prod (yarn, cluster, http://rm.prod.example.com:8088)
# • app_name: wordcount-books
# • script: <script_path>
# • queue: research
# • resources: 2G memory × 2 cores × 2 executors
# This will invoke spark-submit and start a real YARN application.
# Confirm? (y / n)"
# User: "yes, go"
# ─── only NOW may you call confirm_submit_job ───
# 5. Confirm (NOW spark-submit runs)
result = confirm_submit_job(pending_id=pending_id)
job_id = result["job_id"] # STORE
application_id = result["application_id"] # STORE
tracking_url = result["tracking_url"] # STORE
# 6. Poll
import time
while True:
st = get_job_status(job_id=job_id).state
if st in ("SUCCEEDED", "FAILED", "KILLED", "FINISHED"):
break
time.sleep(30)
# 7. Final result
res = get_job_result(job_id=job_id)
print(res.final_status, res.diagnostics)
# If something went wrong and you want to see why:
# logs = get_job_logs(job_id=job_id, tail_chars=20000)
# (If you need to kill the job: ASK THE USER FIRST, then call
# kill_job(job_id=job_id). Don't kill running jobs without an explicit
# "kill it" — the user may have wanted to let it finish.)
```
---
## 10. Reference paths (for the curious)
This skill is the **operation** guide. The architectural / implementation
details live in:
- `CLAUDE.md` — project conventions, persistence layout, two-step submit
rationale, MCP arg pattern, MCP session affinity caveat.
- `docs/superpowers/plans/2026-06-24-spark-executor-mcp.md` — the
original implementation plan with Stage 1 / 2 / 3 breakdown.
- `spark_executor/server.py` — the route + exception-handler layer.
- `spark_executor/tools/requests.py` — the Pydantic body models; useful
to read when the MCP 422 response confuses you.
- `spark_executor/core/job_store.py` — Job persistence; explains the
dual-ID contract and the file-backed JSON layout.
+113 -142
View File
@@ -1,169 +1,140 @@
---
name: sql-context-builder
description: Use when assembling the single trusted SQL context that the SQL generator and reviewer will consume. Triggers include "build SQL context", "固化 join key", "table aliases", "列出来自哪张表", "准备写 SQL", "context 已经齐了". Consumes a validated Logic Plan + Field Mapping and produces a normalized SQL Context with stable aliases, frozen join keys, and explicit fact/dimension tagging. The SQL generator MUST NOT re-derive any of this.
description: 当逻辑计划已经完成,需要固定 SQL 生成和 SQL review 共同依赖的 alias、字段来源、join key、指标、维度、过滤、时间窗口、事实表/维表上下文时使用。消费 logic_plan 和 validation_result,不写 SQL,不调用 MCP。
---
# SQL Context Builder
# SQL Context BuilderSQL 上下文构建)
## Overview
## 目标
Freeze every decision that the SQL generator would otherwise re-invent. Aliases, join keys, field sources, types, role tags (fact vs dimension vs filter), and time columns are all pinned here exactly once. The output is the **single trusted input** for `sql-review` and for any SQL-emitting step.
把 SQL 生成时可能被临时猜测的内容提前固定下来。下游只能使用本 skill 输出的 alias、字段、join、指标、过滤、时间窗口和输出列。
Core principle: **anything not in the SQL Context does not exist for downstream skills.** If a downstream skill needs something new, the request must go back through `metadata-validator` and `logic-planner` first.
## 输入
## When to Use
Use when:
- `logic-planner` returned status `PLANNED`
- The user is about to write Spark SQL and you need a normalized context (aliases, join keys, column sources)
- Multiple SQL drafts need to be generated against the *same* logic — context guarantees they stay consistent
Do NOT use when:
- `logic-plan.status` is `NEED_USER_CONFIRMATION` — go back to `logic-planner`
- The SQL is trivial (one table, no join) — you can still build the context, but skip if the user explicitly opts out
- The task is to translate the plan into SQL — that is a separate concern from this skill
## Inputs
1. `logic_plan` from `logic-planner` (status `PLANNED`)
2. `field_mapping` from `metadata-validator` (status `VALIDATED`)
3. Original `validation_result` for table-level metadata (types, sources)
## Workflow
```dot
digraph sql_context_builder {
"Read logic_plan + field_mapping + validation_result" [shape=box];
"Assign stable alias to every table" [shape=box];
"Freeze join keys (left_alias.field = right_alias.field)" [shape=box];
"Resolve every field to (alias, column, type, source_file)" [shape=box];
"Tag each field: dimension | metric | filter | time | join_key" [shape=box];
"Mark fact vs dimension tables" [shape=box];
"Verify Logic Plan fields all exist in context" [shape=box];
"All required fields present?" [shape=diamond];
"Emit SQL Context" [shape=box];
"Emit SQL Context + missing_fields" [shape=box];
"Read logic_plan + field_mapping + validation_result" -> "Assign stable alias to every table";
"Assign stable alias to every table" -> "Freeze join keys (left_alias.field = right_alias.field)";
"Freeze join keys (left_alias.field = right_alias.field)" -> "Resolve every field to (alias, column, type, source_file)";
"Resolve every field to (alias, column, type, source_file)" -> "Tag each field: dimension | metric | filter | time | join_key";
"Tag each field: dimension | metric | filter | time | join_key" -> "Mark fact vs dimension tables";
"Mark fact vs dimension tables" -> "Verify Logic Plan fields all exist in context";
"Verify Logic Plan fields all exist in context" -> "All required fields present?";
"All required fields present?" -> "Emit SQL Context" [label="yes"];
"All required fields present?" -> "Emit SQL Context + missing_fields" [label="no"];
}
```yaml
logic_plan:
status: PLANNED
validation_result:
status: VALIDATED
field_mapping: {}
joins: []
```
## Rules
## 工作顺序
### Alias Assignment
- One alias per table, never reused.
- Fact table (the one with the primary metric) gets a short alias first (`o`, `f`, `t1`).
- Dimension tables get meaningful aliases (`c` for customer, `p` for product, `ch` for channel).
- Aliases are **frozen** here. The SQL generator and reviewer must not invent new ones.
1. 给每张表分配唯一、稳定、短小的 alias。
2. 标记事实表、维表、码表和目标表。
3. 把每个业务字段解析成 `alias.column`
4. 固定 join 谓词、join 类型和基数假设。
5. 固定指标表达式、维度字段、时间字段、分区字段和过滤字段。
6. 固定时间窗口 SQL 片段,尤其是 T-1 完整日边界。
7. 固定输出列顺序和别名。
8. 检查 logic plan 中引用的字段都能在 context 中找到。
9. 对派生字段记录来源步骤和依赖字段。
### Join Key Freezing
- For every `logic_plan.steps[kind=join]`, freeze the exact predicate:
- `left_alias.col1 = right_alias.col2`
- Include join type (inner / left / right / full / semi / anti).
- Include cardinality assumption carried over from the Logic Plan.
- If two tables could join on multiple keys, freeze the chosen one here. Other candidates are out of scope for SQL.
## 时间窗口上下文
### Field Resolution
For every field used in the Logic Plan, record:
- `alias` (which table it lives on)
- `column` (physical name)
- `type` (from `validation_result`)
- `source_file` (which metadata file it came from — for traceability)
- `role`: one of `dimension` | `metric` | `filter` | `time` | `join_key` | `derived`
当存在时间过滤时,必须输出 `time_windows`。对“近 N 天按 T-1 完整日”:
A field may have multiple roles (e.g. a `customer_id` that is both `join_key` and `dimension`); list all.
```yaml
time_windows:
- name: t_minus_1_complete_days
event_field: fl.login_time
lower_bound_sql: date_sub(current_date(), N)
upper_bound_sql: current_date()
predicate_sql: fl.login_time >= date_sub(current_date(), N) AND fl.login_time < current_date()
boundary: left_closed_right_open
```
### Fact vs Dimension Tagging
- The table containing the primary event/measure is `fact`.
- Tables that enrich (customer, product, channel, city) are `dimension`.
- One fact per query. If the Logic Plan implies multiple facts, raise a `pending_question` — that is a planning issue, not a context issue.
下游 SQL 必须复用 `predicate_sql`,不得临时改成 `BETWEEN`
### Coverage Check
- Every step in `logic_plan.steps` must reference fields that exist in this context.
- Every `field_mapping` entry must appear in this context.
- Every `group_by` and `order_by` field must be tagged.
- Every metric in `aggregate` must have a `metric` tag.
- Every predicate in `filter` must reference a `filter` or `time` tag.
## 转化指标上下文
## Output Contract
转化率必须固定分母、分子和顺序约束:
```yaml
metrics:
- name: conversion_rate
numerator: converted_user_cnt
denominator: new_user_cnt
order_constraint: pay_time >= register_time
null_policy: denominator_zero_returns_0
exists_after_anchor_event: true
```
## 输出格式
```yaml
sql_context:
status: CONTEXT_READY | NEED_USER_CONFIRMATION
aliases:
- alias: o
table: loan_order
role: fact
source_file: metadata/loan_order.json
- alias: c
table: customer_info
role: dimension
source_file: metadata/customer_info.md
joins:
- left: { alias: o, column: customer_id }
right: { alias: c, column: customer_id }
type: left
cardinality_assumption: 1:N
- alias: fl
table: user_login
role: fact | dimension | lookup | target
source_file: "元数据来源"
fields:
- alias: o
column: customer_id
type: string
role: [join_key]
source_file: metadata/loan_order.json
- alias: o
column: loan_amount
type: decimal(18,2)
role: [metric]
source_file: metadata/loan_order.json
- alias: o
column: apply_time
type: timestamp
role: [time, filter]
source_file: metadata/loan_order.json
- alias: c
column: city
type: string
role: [dimension]
source_file: metadata/customer_info.md
metrics:
- alias: total_loan
expr: "SUM(o.loan_amount)"
agg: sum
- alias: customer_cnt
expr: "COUNT(DISTINCT o.customer_id)"
agg: count_distinct
dimensions: ["c.city"]
time_grain:
column: o.apply_time
granularity: day
filters:
- "o.status = 'SUCCESS'"
missing_fields: []
status: <READY | NEED_USER_CONFIRMATION>
- id: fl.user_id
alias: fl
table: user_login
column: user_id
type: bigint
role: [join_key, metric]
nullable: false
source_file: "元数据来源"
joins:
- left: fl.user_id
right: ur.user_id
type: left
cardinality: "N:1"
source_step: S3
time_windows: []
metrics: []
dimensions: []
filters: []
outputs: []
write_intent:
mode: query_only | insert_into | insert_overwrite
target_table: "可为空"
overwrite_confirmed: false
unresolved: []
```
## Forbidden Behaviors
## 用户可见摘要
| Rationalization | Reality |
|---|---|
| "I'll let the SQL generator pick aliases" | Aliases are frozen here. Re-aliasing downstream breaks `sql-review`. |
| "The join key is obvious, skip the freeze" | The whole point of this skill is to freeze join keys. |
| "Field type doesn't matter, it's all strings in SQL" | Type drives aggregations, casts, and partition-pruning decisions. |
| "Derived columns can be invented in SQL" | Every derived column must already appear in the Logic Plan. |
| "Multiple facts? Just join them all" | Multi-fact queries are a Logic Plan issue. Surface and stop. |
除了 YAML,必须输出:
## Completion Criteria
```markdown
## SQL 上下文
### 表别名
| Alias | 表名 | 角色 |
|---|---|---|
`status: READY` only when:
- Every Logic Plan step has matching entries in this context
- `missing_fields` is empty
- Every metric has a defined `agg` and `expr`
- Every join has a frozen key and a type
- The time grain is set
### 字段来源
| 字段 | 来源 | 用途 |
|---|---|---|
### Join 关系
| 左字段 | 右字段 | 类型 | 基数风险 |
|---|---|---|---|
### 指标公式
| 指标 | 分子 | 分母 | 空值策略 |
|---|---|---|---|
### 时间窗口
| 字段 | 下界 | 上界 | 边界 |
|---|---|---|---|
```
## 规则
- alias 一旦生成,下游不能重新命名。
- context 中不存在的字段,下游视为不存在。
- join key 只能来自元数据校验和逻辑计划。
- 指标表达式必须能追溯到 logic plan 的 aggregate 步骤。
- 输出列必须来自 `fields``metrics` 或明确派生字段。
- 如果 SQL 生成需要新字段,回到 metadata-validator 或 logic-planner。
## 停止条件
只在 alias、字段、join、指标、时间字段、过滤字段、输出列和写入意图都完整时输出 `CONTEXT_READY`
+94 -102
View File
@@ -1,124 +1,116 @@
---
name: sql-review
description: Use when performing a static review of a generated PySpark SQL string before it is executed. Triggers include "review this SQL", "SQL 评审", "静态审查", "risk level", "check join cardinality", "数据膨胀", "重复聚合", "分区裁剪". Consumes a SQL string plus the SQL Context that produced it, and emits a risk-graded review (HIGH / MEDIUM / LOW) with concrete fixes. Does NOT modify the SQL — it reports findings.
description: 当一条 Spark SQL 已经生成、通过 guardrails 并准备展示或执行前使用。它基于 sql_context 静态检查字段引用、alias、join key、数据膨胀、重复聚合、时间过滤、转化顺序、GROUP BY、空值、类型和写入风险,只报告问题,不直接改 SQL。
---
# SQL Review
# SQL ReviewSQL 静态评审)
## Overview
## 目标
Static, pre-execution review of a generated PySpark SQL string. The reviewer does **not** rewrite SQL — it emits findings, severity, and suggested fixes. The author of the SQL decides what to act on.
在 SQL 执行前做质量检查,发现可能导致数字错误、执行失败或性能风险的问题。评审只报告风险和建议,不负责直接改写 SQL。
Core principle: **review against the SQL Context, not against vibes.** Every field, alias, and join key referenced in the SQL must exist in the SQL Context that produced it. If it does not, that is itself a `HIGH` finding (the SQL bypassed the contract).
## 输入
## When to Use
Use when:
- A SQL string has been generated from a `sql_context` and is about to be executed
- The user pastes a SQL string and asks for a review, optimization, or risk assessment
- A code change adds a new `spark.sql(...)` call to a PySpark script
Do NOT use when:
- The SQL is not Spark SQL (e.g. plain Postgres, HiveQL outside Spark) — at minimum replace Spark-specific checks with the relevant engine's equivalents, but flag it
- The SQL is a one-off interactive query that won't be reused — the user can opt out
## Inputs
1. The **SQL string** to review
2. The **SQL Context** that the SQL was generated from (or, if absent, an explicit acknowledgement that the SQL is "untrusted / ad-hoc")
If the SQL Context is missing, raise a single `HIGH` finding: *"SQL has no associated SQL Context; downstream skills cannot guarantee field/join correctness."* Then continue with structural checks.
## Severity Model
| Level | Meaning | Action |
|---|---|---|
| `HIGH` | Will produce wrong numbers, fail to run, or violate policy | Block execution. Must fix. |
| `MEDIUM` | Likely correct, but inefficient, fragile, or risky at scale | Fix before production. OK in dev. |
| `LOW` | Style / best-practice nit | Optional fix. |
Emit findings in this order: `HIGH``MEDIUM``LOW`. A single `HIGH` finding makes the overall review verdict `FAIL`.
## Workflow
```dot
digraph sql_review {
"Parse SQL (logical, not just regex)" [shape=box];
"Cross-check aliases, columns, join keys against SQL Context" [shape=box];
"Detect multi-statement or forbidden verbs (guardrail)" [shape=box];
"Analyze join cardinality & inflation risk" [shape=box];
"Detect repeated aggregation on same measure" [shape=box];
"Detect full table scan & missing partition filter" [shape=box];
"Verify GROUP BY matches SELECT" [shape=box];
"Check null handling on join keys & measures" [shape=box];
"Check field types vs operations" [shape=box];
"Aggregate findings into verdict" [shape=box];
"Parse SQL (logical, not just regex)" -> "Cross-check aliases, columns, join keys against SQL Context";
"Cross-check aliases, columns, join keys against SQL Context" -> "Detect multi-statement or forbidden verbs (guardrail)";
"Detect multi-statement or forbidden verbs (guardrail)" -> "Analyze join cardinality & inflation risk";
"Analyze join cardinality & inflation risk" -> "Detect repeated aggregation on same measure";
"Detect repeated aggregation on same measure" -> "Detect full table scan & missing partition filter";
"Detect full table scan & missing partition filter" -> "Verify GROUP BY matches SELECT";
"Verify GROUP BY matches SELECT" -> "Check null handling on join keys & measures";
"Check null handling on join keys & measures" -> "Check field types vs operations";
"Check field types vs operations" -> "Aggregate findings into verdict";
}
```yaml
sql: "待评审的 Spark SQL"
sql_context:
status: CONTEXT_READY
sql_guard:
status: PASS
logic_plan:
status: PLANNED
```
## Checks (run all)
如果没有 `sql_context`,也可以继续做基础结构检查,但必须报告一个 HIGH 风险:缺少上下文,无法保证字段和 join 正确。
1. **Reference integrity** — every `<alias>.<column>` in the SQL must exist in the SQL Context. Mismatches are `HIGH`.
2. **Join key match** — every `ON` predicate must match a frozen join from the SQL Context. Invented joins are `HIGH`.
3. **Forbidden verbs / multi-statement** — defer to `pyspark-sql-guardrails`. Any DDL/DML/maintenance verb, or any `;`-stacked statement, is `HIGH`.
4. **Join cardinality risk** — for every join, if the SQL Context marks cardinality as `1:N` or `N:N` and there is no dedup, raise `MEDIUM` (1:N) or `HIGH` (N:N).
5. **Repeated aggregation** — if the same measure is aggregated twice (e.g. once in a subquery, once outside) without intentional re-aggregation, raise `HIGH`.
6. **Partition / time filter** — for any fact table with a known time field, missing `WHERE` on that field is `HIGH` (full scan risk).
7. **GROUP BY completeness** — every selected non-aggregated column must be in `GROUP BY`. Mismatch is `HIGH`.
8. **Null handling** — nullable join keys with `INNER JOIN` silently drop rows (`MEDIUM`); nullable measures aggregated without `COALESCE` can produce `NULL` that breaks downstream `WHERE ... > 0` (`MEDIUM`).
9. **Type mismatch** — string columns used in numeric aggregation without cast (`MEDIUM`); date string compared to `TIMESTAMP` literal (`MEDIUM`).
10. **Spark best practice**`COUNT(*)` vs `COUNT(1)` is a style nit (`LOW`); `SELECT *` in production is `MEDIUM`; missing `DISTRIBUTE BY` / `CLUSTER BY` for skewed joins is `LOW` to `MEDIUM`.
## 检查项
## Output Contract
1. 字段引用:每个 `alias.column` 是否存在于 `sql_context.fields`
2. aliasSQL 是否发明了 context 中没有的 alias。
3. join keyON 条件是否匹配 context 中固定的 join。
4. join 基数:1:N 或 N:N join 是否会造成数据膨胀。
5. 指标公式:聚合表达式是否符合 logic plan 和 context。
6. 重复聚合:同一指标是否被重复 sum/count。
7. 时间过滤:事实表是否缺少必要时间或分区过滤。
8. 时间边界:T-1 完整日是否使用左闭右开窗口。
9. 转化顺序:后置事件是否满足前置事件之后发生。
10. GROUP BY:非聚合输出列是否都在 group by 中。
11. 空值策略:join key 和度量字段是否需要 coalesce 或保留策略。
12. 类型匹配:字符串日期、字符串金额是否需要 cast。
13. 写入风险:INSERT 目标列、overwrite 确认、分区覆盖范围是否明确。
14. 执行成本:是否存在无分区裁剪的大表全扫、笛卡尔积、窗口无 partition。
## 时间边界检查
当需求为“近 N 天按 T-1 完整日”且字段是 timestamp/date-like
推荐:
```sql
event_time >= date_sub(current_date(), N)
AND event_time < current_date()
```
风险写法:
```sql
event_time BETWEEN date_add(current_date(), -N) AND date_add(current_date(), -1)
```
如果 SQL 使用风险写法,至少标为 `MEDIUM`;如果会明显漏掉结束日期白天数据,标为 `HIGH`
## 转化顺序检查
对注册、付费、激活、下单等漏斗指标,必须检查后置事件是否在前置事件之后。若口径是“注册后付费”,SQL 必须有等价约束:
```sql
pay_time >= register_time
```
如果 SQL 先按用户取 `MIN(pay_time)`,再和 `register_time` 比较,而没有在聚合前加入 `pay_time >= register_time`,标为 `HIGH``MEDIUM`,因为这可能漏算注册后仍然完成付费的用户。
## 输出格式
```yaml
sql_review:
verdict: <PASS | FAIL>
highest_severity: <HIGH | MEDIUM | LOW>
summary: "<one-sentence summary>"
verdict: PASS | FAIL
highest_severity: NONE | LOW | MEDIUM | HIGH
summary: "一句话结论"
findings:
- id: F1
severity: HIGH
check: reference_integrity
location: "<line/column or SQL fragment>"
message: "<what is wrong>"
fix: "<concrete suggested change>"
- id: F2
severity: MEDIUM
check: join_cardinality
location: "JOIN customer_info"
message: "1:N join without pre-aggregation; risk of row inflation"
fix: "Aggregate loan_order by customer_id before joining"
- id: R1
severity: HIGH | MEDIUM | LOW
check: reference_integrity | join_cardinality | metric_formula | time_filter | time_boundary | conversion_order | group_by | type_check | write_risk | performance | other
location: "SQL 片段或行号"
message: "问题描述"
recommendation: "建议修复方向"
notes:
- "SQL Context provided: yes (sql_context.aliases = [...])"
- "Reviewed by: sql-review v1"
- "评审假设"
```
## Forbidden Behaviors
## 用户可见摘要
| Rationalization | Reality |
|---|---|
| "The SQL looks fine, no need to cross-check the context" | A column can look fine and still not exist in the source table. Cross-check is mandatory. |
| "I'll fix the SQL while reviewing" | The reviewer reports. The author fixes. Separation of concerns. |
| "No SQL Context was provided, so I'll skip field checks" | Missing context is itself a `HIGH` finding. Never silently skip. |
| "It's just a dev query, skip partition check" | The check is cheap. Skipping it produces silent full scans. |
| "1:N joins are normal, no risk" | 1:N is fine *if* the grain is preserved. The reviewer must verify, not assume. |
| "Regex is enough to check GROUP BY" | GROUP BY membership is structural. Use a parser, not regex. |
除了 YAML,必须输出:
## Completion Criteria
```markdown
## SQL Review
- 结论:PASS / FAIL
- 最高风险:NONE / LOW / MEDIUM / HIGH
- 已检查:字段引用、join、GROUP BY、时间窗口、转化顺序、空值、类型、性能
- 主要风险:...
- 是否可交付:...
```
A review is complete when:
- All 10 checks have been evaluated
- Every `HIGH` finding has a `fix`
- `verdict` is `PASS` iff `highest_severity` is `MEDIUM` or `LOW`
- The output YAML can be pasted into a PR comment without edits
## 严重程度
- `HIGH`:会导致数字错误、执行失败、越权写入或违反策略,必须阻断。
- `MEDIUM`:大概率可跑,但有性能、稳定性或边界风险,需要用户接受。
- `LOW`:风格或可维护性问题。
只要存在 HIGH`verdict` 就是 `FAIL`。只有 LOW/MEDIUM 时,可以把风险交给用户决定是否接受。
## 禁止行为
- 不要直接改 SQL;只输出发现和建议。
- 不要凭直觉审字段,必须对照 `sql_context`
- 不要因为 SQL 能通过 guardrails 就跳过业务正确性审查。
- 不要忽略写入模式,尤其是 `INSERT OVERWRITE`