---

name: "snowflake-development" description: "Use when writing Snowflake SQL, building data pipelines with Dynamic Tables or Streams/Tasks, using Cortex AI functions, creating Cortex Agents, writing Snowpark Python, configuring dbt for Snowflake, or troubleshooting Snowflake errors."

Snowflake Development

Snowflake SQL, data pipelines, Cortex AI, and Snowpark Python development. Covers the colon-prefix rule, semi-structured data, MERGE upserts, Dynamic Tables, Streams+Tasks, Cortex AI functions, agent specs, performance tuning, and security hardening.

Originally contributed by James Cha-Earley — enhanced and integrated by the claude-skills team.

Quick Start

# Generate a MERGE upsert template
python scripts/snowflake_query_helper.py merge --target customers --source staging_customers --key customer_id --columns name,email,updated_at

# Generate a Dynamic Table template
python scripts/snowflake_query_helper.py dynamic-table --name cleaned_events --warehouse transform_wh --lag "5 minutes"

# Generate RBAC grant statements
python scripts/snowflake_query_helper.py grant --role analyst_role --database analytics --schemas public,staging --privileges SELECT,USAGE

---

SQL Best Practices

Naming and Style

• Use snake_case for all identifiers. Avoid double-quoted identifiers -- they force case-sensitive names that require constant quoting.
• Use CTEs (WITH clauses) over nested subqueries.
• Use CREATE OR REPLACE for idempotent DDL.
• Use explicit column lists -- never SELECT * in production. Snowflake's columnar storage scans only referenced columns, so explicit lists reduce I/O.

Stored Procedures -- Colon Prefix Rule

In SQL stored procedures (BEGIN...END blocks), variables and parameters must use the colon : prefix inside SQL statements. Without it, Snowflake treats them as column identifiers and raises "invalid identifier" errors.

-- WRONG: missing colon prefix
SELECT name INTO result FROM users WHERE id = p_id;

-- CORRECT: colon prefix on both variable and parameter
SELECT name INTO :result FROM users WHERE id = :p_id;

This applies to DECLARE variables, LET variables, and procedure parameters when used inside SELECT, INSERT, UPDATE, DELETE, or MERGE.

Semi-Structured Data

• VARIANT, OBJECT, ARRAY for JSON/Avro/Parquet/ORC.
• Access nested fields: src:customer.name::STRING. Always cast with ::TYPE.
• VARIANT null vs SQL NULL: JSON null is stored as the string "null". Use STRIP_NULL_VALUE = TRUE on load.
• Flatten arrays: SELECT f.value:name::STRING FROM my_table, LATERAL FLATTEN(input => src:items) f;

MERGE for Upserts

MERGE INTO target t USING source s ON t.id = s.id
WHEN MATCHED THEN UPDATE SET t.name = s.name, t.updated_at = CURRENT_TIMESTAMP()
WHEN NOT MATCHED THEN INSERT (id, name, updated_at) VALUES (s.id, s.name, CURRENT_TIMESTAMP());

See references/snowflake_sql_and_pipelines.md for deeper SQL patterns and anti-patterns.

---

Data Pipelines

Choosing Your Approach

| Approach | When to Use | |----------|-------------| | Dynamic Tables | Declarative transformations. Default choice. Define the query, Snowflake handles refresh. | | Streams + Tasks | Imperative CDC. Use for procedural logic, stored procedure calls, complex branching. | | Snowpipe | Continuous file loading from cloud storage (S3, GCS, Azure). |

Dynamic Tables

CREATE OR REPLACE DYNAMIC TABLE cleaned_events
    TARGET_LAG = '5 minutes'
    WAREHOUSE = transform_wh
    AS
    SELECT event_id, event_type, user_id, event_timestamp
    FROM raw_events
    WHERE event_type IS NOT NULL;

Key rules:

• Set TARGET_LAG progressively: tighter at the top of the DAG, looser downstream.
• Incremental DTs cannot depend on Full-refresh DTs.
• SELECT * breaks on upstream schema changes -- use explicit column lists.
• Views cannot sit between two Dynamic Tables in the DAG.

Streams and Tasks

CREATE OR REPLACE STREAM raw_stream ON TABLE raw_events;

CREATE OR REPLACE TASK process_events
    WAREHOUSE = transform_wh
    SCHEDULE = 'USING CRON 0 */1 * * * America/Los_Angeles'
    WHEN SYSTEM$STREAM_HAS_DATA('raw_stream')
    AS INSERT INTO cleaned_events SELECT ... FROM raw_stream;

-- Tasks start SUSPENDED. You MUST resume them.
ALTER TASK process_events RESUME;

See references/snowflake_sql_and_pipelines.md for DT debugging queries and Snowpipe patterns.

---

Cortex AI

Function Reference

| Function | Purpose | |----------|---------| | AI_COMPLETE | LLM completion (text, images, documents) | | AI_CLASSIFY | Classify text into categories (up to 500 labels) | | AI_FILTER | Boolean filter on text or images | | AI_EXTRACT | Structured extraction from text/images/documents | | AI_SENTIMENT | Sentiment score (-1 to 1) | | AI_PARSE_DOCUMENT | OCR or layout extraction from documents | | AI_REDACT | PII removal from text |

Deprecated names (do NOT use): COMPLETE, CLASSIFY_TEXT, EXTRACT_ANSWER, PARSE_DOCUMENT, SUMMARIZE, TRANSLATE, SENTIMENT, EMBED_TEXT_768.

TO_FILE -- Common Pitfall

Stage path and filename are separate arguments:

-- WRONG: single combined argument
TO_FILE('@stage/file.pdf')

-- CORRECT: two arguments
TO_FILE('@db.schema.mystage', 'invoice.pdf')

Cortex Agents

Agent specs use a JSON structure with top-level keys: models, instructions, tools, tool_resources.

• Use $spec$ delimiter (not $$).
• models must be an object, not an array.
• tool_resources is a separate top-level key, not nested inside tools.
• Tool descriptions are the single biggest factor in agent quality.

See references/cortex_ai_and_agents.md for full agent spec examples and Cortex Search patterns.

---

Snowpark Python

from snowflake.snowpark import Session
import os

session = Session.builder.configs({
    "account": os.environ["SNOWFLAKE_ACCOUNT"],
    "user": os.environ["SNOWFLAKE_USER"],
    "password": os.environ["SNOWFLAKE_PASSWORD"],
    "role": "my_role", "warehouse": "my_wh",
    "database": "my_db", "schema": "my_schema"
}).create()

• Never hardcode credentials. Use environment variables or key pair auth.
• DataFrames are lazy -- executed on collect() / show().
• Do NOT call collect() on large DataFrames. Process server-side with DataFrame operations.
• Use vectorized UDFs (10-100x faster) for batch and ML workloads.

dbt on Snowflake

-- Dynamic table materialization (streaming/near-real-time marts):
{{ config(materialized='dynamic_table', snowflake_warehouse='transforming', target_lag='1 hour') }}

-- Incremental materialization (large fact tables):
{{ config(materialized='incremental', unique_key='event_id') }}

-- Snowflake-specific configs (combine with any materialization):
{{ config(transient=true, copy_grants=true, query_tag='team_daily') }}

• Do NOT use {{ this }} without {% if is_incremental() %} guard.
• Use dynamic_table materialization for streaming or near-real-time marts.

Performance

• Cluster keys: Only for multi-TB tables. Apply on WHERE / JOIN / GROUP BY columns.
• Search Optimization: ALTER TABLE t ADD SEARCH OPTIMIZATION ON EQUALITY(col);
• Warehouse sizing: Start X-Small, scale up. Set AUTO_SUSPEND = 60, AUTO_RESUME = TRUE.
• Separate warehouses per workload (load, transform, query).

Security

• Follow least-privilege RBAC. Use database roles for object-level grants.
• Audit ACCOUNTADMIN regularly: SHOW GRANTS OF ROLE ACCOUNTADMIN;
• Use network policies for IP allowlisting.
• Use masking policies for PII columns and row access policies for multi-tenant isolation.

---

Proactive Triggers

Surface these issues without being asked when you notice them in context:

• Missing colon prefix in SQL stored procedures -- flag immediately, this causes "invalid identifier" at runtime.
• SELECT * in Dynamic Tables -- flag as a schema-change time bomb.
• Deprecated Cortex function names (CLASSIFY_TEXT, SUMMARIZE, etc.) -- suggest the current AI_* equivalents.
• Task not resumed after creation -- remind that tasks start SUSPENDED.
• Hardcoded credentials in Snowpark code -- flag as a security risk.

---

Common Errors

| Error | Cause | Fix | |-------|-------|-----| | "Object does not exist" | Wrong database/schema context or missing grants | Fully qualify names (db.schema.table), check grants | | "Invalid identifier" in procedure | Missing colon prefix on variable | Use :variable_name inside SQL statements | | "Numeric value not recognized" | VARIANT field not cast | Cast explicitly: src:field::NUMBER(10,2) | | Task not running | Forgot to resume after creation | ALTER TASK task_name RESUME; | | DT refresh failing | Schema change upstream or tracking disabled | Use explicit columns, verify change tracking | | TO_FILE error | Combined path as single argument | Split into two args: TO_FILE('@stage', 'file.pdf') |

---

Practical Workflows

Workflow 1: Build a Reporting Pipeline (30 min)

1. Stage raw data: Create external stage pointing to S3/GCS/Azure, set up Snowpipe for auto-ingest
2. Clean with Dynamic Table: Create DT with TARGET_LAG = '5 minutes' that filters nulls, casts types, deduplicates
3. Aggregate with downstream DT: Second DT that joins cleaned data with dimension tables, computes metrics
4. Expose via Secure View: Create SECURE VIEW for the BI tool / API layer
5. Grant access: Use snowflake_query_helper.py grant to generate RBAC statements

Workflow 2: Add AI Classification to Existing Data

1. Identify the column: Find the text column to classify (e.g., support tickets, reviews)
2. Test with AI_CLASSIFY: SELECT AI_CLASSIFY(text_col, ['bug', 'feature', 'question']) FROM table LIMIT 10;
3. Create enrichment DT: Dynamic Table that runs AI_CLASSIFY on new rows automatically
4. Monitor costs: Cortex AI is billed per token — sample before running on full tables

Workflow 3: Debug a Failing Pipeline

1. Check task history: SELECT * FROM TABLE(INFORMATION_SCHEMA.TASK_HISTORY()) WHERE STATE = 'FAILED' ORDER BY SCHEDULED_TIME DESC;
2. Check DT refresh: SELECT * FROM TABLE(INFORMATION_SCHEMA.DYNAMIC_TABLE_REFRESH_HISTORY('my_dt')) ORDER BY REFRESH_END_TIME DESC;
3. Check stream staleness: SHOW STREAMS; -- check stale_after column
4. Consult troubleshooting reference: See references/troubleshooting.md for error-specific fixes

---

Anti-Patterns

| Anti-Pattern | Why It Fails | Better Approach | |---|---|---| | SELECT * in Dynamic Tables | Schema changes upstream break the DT silently | Use explicit column lists | | Missing colon prefix in procedures | "Invalid identifier" runtime error | Always use :variable_name in SQL blocks | | Single warehouse for all workloads | Contention between load, transform, and query | Separate warehouses per workload type | | Hardcoded credentials in Snowpark | Security risk, breaks in CI/CD | Use os.environ[] or key pair auth | | collect() on large DataFrames | Pulls entire result set to client memory | Process server-side with DataFrame operations | | Nested subqueries instead of CTEs | Unreadable, hard to debug, Snowflake optimizes CTEs better | Use WITH clauses | | Using deprecated Cortex functions | CLASSIFY_TEXT, SUMMARIZE etc. will be removed | Use AI_CLASSIFY, AI_COMPLETE etc. | | Tasks without WHEN SYSTEM$STREAM_HAS_DATA | Task runs on schedule even with no new data, wasting credits | Add the WHEN clause for stream-driven tasks | | Double-quoted identifiers | Forces case-sensitive names across all queries | Use snake_case unquoted identifiers |

---

Cross-References

| Skill | Relationship | |-------|-------------| | engineering/sql-database-assistant | General SQL patterns — use for non-Snowflake databases | | engineering/database-designer | Schema design — use for data modeling before Snowflake implementation | | engineering-team/senior-data-engineer | Broader data engineering — pipelines, Spark, Airflow, data quality | | engineering-team/senior-data-scientist | Analytics and ML — use alongside Snowpark for feature engineering | | engineering-team/senior-devops | CI/CD for Snowflake deployments (Terraform, GitHub Actions) |

---

Reference Documentation

| Document | Contents | |----------|----------| | references/snowflake_sql_and_pipelines.md | SQL patterns, MERGE templates, Dynamic Table debugging, Snowpipe, anti-patterns | | references/cortex_ai_and_agents.md | Cortex AI functions, agent spec structure, Cortex Search, Snowpark | | references/troubleshooting.md | Error reference, debugging queries, common fixes |