❄️ Snowflake Cheat Sheet

CREATE WAREHOUSE etl_wh
  WITH WAREHOUSE_SIZE = 'MEDIUM'
  AUTO_SUSPEND = 300
  AUTO_RESUME = TRUE
  INITIALLY_SUSPENDED = TRUE
  MIN_CLUSTER_COUNT = 1
  MAX_CLUSTER_COUNT = 3
  SCALING_POLICY = 'STANDARD';

-- Sizes: XSMALL, SMALL, MEDIUM, LARGE, XLARGE, 2XL, 3XL, 4XL
-- Each size doubles compute & cost

ALTER WAREHOUSE etl_wh SET WAREHOUSE_SIZE = 'LARGE';
ALTER WAREHOUSE etl_wh SUSPEND;
ALTER WAREHOUSE etl_wh RESUME;
ALTER WAREHOUSE etl_wh SET AUTO_SUSPEND = 60;

SHOW WAREHOUSES;
DROP WAREHOUSE IF EXISTS temp_wh;

-- Resource monitor
CREATE RESOURCE MONITOR monthly_limit
  WITH CREDIT_QUOTA = 500
  TRIGGERS
    ON 75 PERCENT DO NOTIFY
    ON 100 PERCENT DO SUSPEND;

CREATE DATABASE analytics;
CREATE DATABASE IF NOT EXISTS staging;
USE DATABASE analytics;

CREATE SCHEMA raw;
CREATE SCHEMA curated;
CREATE SCHEMA analytics.reporting;

SHOW DATABASES;
SHOW SCHEMAS IN DATABASE analytics;

-- Permanent table (default)
CREATE TABLE orders (id INT, amount FLOAT, ts TIMESTAMP);

-- Transient (no Fail-safe, lower cost)
CREATE TRANSIENT TABLE staging_data (...);

-- Temporary (session-scoped, auto-dropped)
CREATE TEMPORARY TABLE temp_results AS
  SELECT * FROM orders WHERE status = 'pending';

-- External table (reads from stage)
CREATE EXTERNAL TABLE ext_data
  WITH LOCATION = @my_stage
  FILE_FORMAT = (TYPE = 'PARQUET');

CREATE SEQUENCE order_seq START = 1 INCREMENT = 1;
SELECT order_seq.NEXTVAL;

-- Tags for governance
CREATE TAG pii_level ALLOWED_VALUES 'HIGH', 'MEDIUM', 'LOW';
ALTER TABLE users SET TAG pii_level = 'HIGH';
ALTER TABLE users MODIFY COLUMN email SET TAG pii_level = 'HIGH';

-- Internal stage
CREATE STAGE my_internal_stage;
PUT file:///tmp/data.csv @my_internal_stage;

-- External S3 stage
CREATE STAGE my_s3_stage
  URL = 's3://bucket/path/'
  STORAGE_INTEGRATION = my_s3_int;

-- External GCS stage
CREATE STAGE my_gcs_stage
  URL = 'gcs://bucket/path/'
  STORAGE_INTEGRATION = my_gcs_int;

-- External Azure stage
CREATE STAGE my_az_stage
  URL = 'azure://account.blob.core.windows.net/container/'
  STORAGE_INTEGRATION = my_az_int;

LIST @my_s3_stage;

CREATE FILE FORMAT csv_fmt
  TYPE = 'CSV'
  FIELD_DELIMITER = ','
  SKIP_HEADER = 1
  NULL_IF = ('NULL', 'null', '')
  FIELD_OPTIONALLY_ENCLOSED_BY = '"'
  ERROR_ON_COLUMN_COUNT_MISMATCH = FALSE;

CREATE FILE FORMAT json_fmt TYPE = 'JSON' STRIP_OUTER_ARRAY = TRUE;
CREATE FILE FORMAT parquet_fmt TYPE = 'PARQUET';
CREATE FILE FORMAT avro_fmt TYPE = 'AVRO';

-- Load from stage
COPY INTO my_table
  FROM @my_s3_stage/data/
  FILE_FORMAT = (FORMAT_NAME = csv_fmt)
  PATTERN = '.*\.csv'
  ON_ERROR = 'CONTINUE';

-- Load with transformations
COPY INTO my_table (id, name, created)
  FROM (
    SELECT $1, UPPER($2), TO_TIMESTAMP($3)
    FROM @my_stage
  )
  FILE_FORMAT = csv_fmt;

-- Unload to stage
COPY INTO @my_stage/export/
  FROM my_table
  FILE_FORMAT = (TYPE = 'PARQUET')
  HEADER = TRUE;

CREATE PIPE my_pipe
  AUTO_INGEST = TRUE
AS
  COPY INTO my_table
  FROM @my_s3_stage
  FILE_FORMAT = csv_fmt;

-- Check pipe status
SELECT SYSTEM$PIPE_STATUS('my_pipe');

-- Refresh manually
ALTER PIPE my_pipe REFRESH;

CREATE STREAM orders_stream ON TABLE raw.orders;

-- Query changes since last DML consumption
SELECT * FROM orders_stream;
-- Metadata columns:
--   METADATA$ACTION   (INSERT / DELETE)
--   METADATA$ISUPDATE (TRUE for UPDATE)
--   METADATA$ROW_ID

-- Append-only stream (inserts only)
CREATE STREAM inserts_only
  ON TABLE raw.events
  APPEND_ONLY = TRUE;

-- Check if stream has data
SELECT SYSTEM$STREAM_HAS_DATA('orders_stream');

-- Scheduled task
CREATE TASK hourly_load
  WAREHOUSE = etl_wh
  SCHEDULE = 'USING CRON 0 * * * * UTC'
  WHEN SYSTEM$STREAM_HAS_DATA('orders_stream')
AS
  INSERT INTO curated.orders
  SELECT * FROM orders_stream WHERE METADATA$ACTION = 'INSERT';

-- Child task (DAG)
CREATE TASK transform_task
  WAREHOUSE = etl_wh
  AFTER hourly_load
AS
  CALL transform_procedure();

-- Manage tasks
ALTER TASK hourly_load RESUME;
ALTER TASK hourly_load SUSPEND;

-- Monitor
SELECT * FROM TABLE(INFORMATION_SCHEMA.TASK_HISTORY())
ORDER BY SCHEDULED_TIME DESC LIMIT 20;

-- Query data as of N seconds ago
SELECT * FROM my_table AT (OFFSET => -3600);

-- Query at specific timestamp
SELECT * FROM my_table AT (TIMESTAMP => '2024-07-01 12:00:00'::TIMESTAMP);

-- Query before a specific statement
SELECT * FROM my_table BEFORE (STATEMENT => '<query_id>');

-- Restore dropped table
UNDROP TABLE accidentally_dropped;
UNDROP SCHEMA dropped_schema;
UNDROP DATABASE dropped_db;

-- Set retention period (Enterprise: up to 90 days)
ALTER TABLE my_table SET DATA_RETENTION_TIME_IN_DAYS = 30;

CREATE TABLE orders_backup CLONE orders;
CREATE SCHEMA dev_schema CLONE prod_schema;
CREATE DATABASE dev_db CLONE prod_db;

-- Clone at a point in time
CREATE TABLE restored CLONE my_table
  AT (TIMESTAMP => '2024-07-01 12:00:00'::TIMESTAMP);

-- Column type for JSON / XML / Avro
CREATE TABLE events (id INT, data VARIANT);

-- Access nested fields
SELECT
  data:user:name::STRING AS user_name,
  data:user:age::INT AS age,
  data:items[0]:product::STRING AS first_product
FROM events;

-- Flatten array to rows
SELECT
  e.id,
  f.value:name::STRING AS item_name,
  f.value:qty::INT AS quantity
FROM events e,
  LATERAL FLATTEN(input => e.data:items) f;

-- Flatten nested object keys
SELECT
  f.key AS metric_name,
  f.value::FLOAT AS metric_value
FROM events e,
  LATERAL FLATTEN(input => e.data:metrics) f;

SELECT PARSE_JSON('{"a":1, "b":2}') AS j;

SELECT OBJECT_CONSTRUCT(
  'name', name,
  'dept', department,
  'salary', salary
) AS json_obj
FROM employees;

SELECT ARRAY_CONSTRUCT(1, 2, 3) AS arr;
SELECT ARRAY_AGG(name) WITHIN GROUP (ORDER BY name) FROM employees;

CREATE OR REPLACE PROCEDURE merge_data(src STRING, tgt STRING)
  RETURNS STRING
  LANGUAGE SQL
AS
BEGIN
  EXECUTE IMMEDIATE 'INSERT INTO ' || tgt || ' SELECT * FROM ' || src;
  RETURN 'Loaded ' || src || ' into ' || tgt;
END;

CREATE OR REPLACE PROCEDURE process_data(table_name STRING)
  RETURNS STRING
  LANGUAGE PYTHON
  RUNTIME_VERSION = '3.11'
  PACKAGES = ('snowflake-snowpark-python')
  HANDLER = 'run'
AS $$
def run(session, table_name):
    df = session.table(table_name)
    result = df.filter(df['status'] == 'active').count()
    return f'Active rows: {result}'
$$;

-- Scalar UDF
CREATE FUNCTION celsius_to_f(c FLOAT)
  RETURNS FLOAT
  LANGUAGE SQL
AS 'c * 9/5 + 32';

-- Python UDF
CREATE FUNCTION classify_salary(salary FLOAT)
  RETURNS STRING
  LANGUAGE PYTHON
  RUNTIME_VERSION = '3.11'
  HANDLER = 'classify'
AS $$
def classify(salary):
    if salary > 100000: return 'Senior'
    elif salary > 60000: return 'Mid'
    return 'Junior'
$$;

SELECT name, classify_salary(salary) FROM employees;

CREATE ROLE analyst_role;
CREATE ROLE etl_role;

GRANT USAGE ON WAREHOUSE etl_wh TO ROLE etl_role;
GRANT USAGE ON DATABASE analytics TO ROLE analyst_role;
GRANT SELECT ON ALL TABLES IN SCHEMA analytics.public TO ROLE analyst_role;
GRANT SELECT ON FUTURE TABLES IN SCHEMA analytics.public TO ROLE analyst_role;

GRANT ROLE analyst_role TO USER mukesh;

-- Role hierarchy
GRANT ROLE analyst_role TO ROLE sysadmin;

-- Provider
CREATE SHARE my_share;
GRANT USAGE ON DATABASE analytics TO SHARE my_share;
GRANT SELECT ON TABLE analytics.public.reports TO SHARE my_share;
ALTER SHARE my_share ADD ACCOUNTS = consumer_account;

-- Consumer
CREATE DATABASE shared_data FROM SHARE provider.my_share;

CREATE DYNAMIC TABLE curated_orders
  TARGET_LAG = '1 hour'
  WAREHOUSE = etl_wh
AS
  SELECT
    o.id, c.name, o.amount, o.order_date
  FROM raw.orders o
  JOIN raw.customers c ON o.customer_id = c.id
  WHERE o.status = 'completed';

-- Snowflake auto-refreshes based on target lag
ALTER DYNAMIC TABLE curated_orders REFRESH;

CREATE MATERIALIZED VIEW mv_daily_revenue AS
  SELECT DATE_TRUNC('day', order_date) AS day,
         SUM(amount) AS revenue
  FROM orders
  GROUP BY 1;

-- Auto-maintained by Snowflake on base table changes
-- Best for: small aggregation result sets with frequent reads

-- Recent queries
SELECT query_id, query_text, execution_status,
       total_elapsed_time/1000 AS seconds,
       bytes_scanned, rows_produced
FROM TABLE(INFORMATION_SCHEMA.QUERY_HISTORY())
ORDER BY start_time DESC LIMIT 20;

-- Expensive queries
SELECT * FROM SNOWFLAKE.ACCOUNT_USAGE.QUERY_HISTORY
WHERE TOTAL_ELAPSED_TIME > 60000  -- > 60 seconds
ORDER BY TOTAL_ELAPSED_TIME DESC;

-- Clustering keys (for large tables)
ALTER TABLE orders CLUSTER BY (order_date, region);

-- Check clustering info
SELECT SYSTEM$CLUSTERING_INFORMATION('orders', '(order_date)');

-- Search optimization
ALTER TABLE orders ADD SEARCH OPTIMIZATION
  ON EQUALITY(customer_id)
  ON SUBSTRING(product_name);