opencode-build/skills/pyspark-sql-guardrails/SKILL.md

name, description

name	description
pyspark-sql-guardrails	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.

PySpark SQL Guardrails

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.

The whole flow is two commands and one condition:

# 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:

# 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):

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.

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.

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:

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:

# 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:

# 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":

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:

PASS - first keyword: select, body length: 420

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:

# 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

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

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).

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.

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.

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)

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.