Snowflake Cheat Sheet

CREATE WAREHOUSE analytics_wh
  WAREHOUSE_SIZE = 'XSMALL'
  AUTO_SUSPEND = 60
  AUTO_RESUME = TRUE
  INITIALLY_SUSPENDED = TRUE;
-- storage is separate from this compute
-- a repeated query can hit result cache without resuming the warehouse

ALTER WAREHOUSE analytics_wh SET WAREHOUSE_SIZE = 'SMALL';

ALTER WAREHOUSE load_wh SET
  MIN_CLUSTER_COUNT = 1
  MAX_CLUSTER_COUNT = 3
  SCALING_POLICY = 'STANDARD';
-- STANDARD adds a cluster quickly; ECONOMY waits to save credits
-- a bigger size speeds one query; more clusters take more concurrent queries

USE WAREHOUSE analytics_wh;
ALTER WAREHOUSE analytics_wh SUSPEND;
-- auto-suspend stops credit use about a minute after the last query

CREATE DATABASE analytics;
CREATE SCHEMA analytics.raw;
CREATE SCHEMA analytics.gold;

CREATE TABLE analytics.raw.orders (
  order_id NUMBER,
  customer_id NUMBER,
  amount NUMBER(12, 2),
  ordered_at TIMESTAMP_NTZ,
  updated_at TIMESTAMP_NTZ
);
CREATE TABLE analytics.gold.orders LIKE analytics.raw.orders;

CREATE TABLE analytics.gold.customers (
  customer_id NUMBER,
  email STRING,
  region STRING
);

CREATE SCHEMA analytics.raw;     -- landed files, lightly typed
CREATE SCHEMA analytics.silver;  -- cleaned and deduped
CREATE SCHEMA analytics.gold;    -- marts people query
-- ELT loads raw first, then transforms inside Snowflake

USE DATABASE analytics;
USE SCHEMA gold;
SELECT order_id FROM analytics.gold.orders;
-- unquoted names are stored uppercase
-- a three-part name still works after USE SCHEMA changes

CREATE ROLE analyst;
CREATE ROLE loader;
CREATE USER avery
  DEFAULT_ROLE = analyst
  DEFAULT_WAREHOUSE = analytics_wh
  DEFAULT_NAMESPACE = analytics.gold;
GRANT ROLE analyst TO USER avery;
-- set the password or key pair outside this script
-- grant roles to users, and privileges to roles

GRANT USAGE ON WAREHOUSE analytics_wh TO ROLE analyst;
GRANT USAGE ON DATABASE analytics TO ROLE analyst;
GRANT USAGE ON SCHEMA analytics.gold TO ROLE analyst;
GRANT SELECT ON ALL TABLES IN SCHEMA analytics.gold TO ROLE analyst;
-- USAGE on the warehouse, database, and schema is required before SELECT
-- loader can insert into raw; analyst cannot

GRANT SELECT ON FUTURE TABLES IN SCHEMA analytics.gold TO ROLE analyst;
GRANT INSERT, SELECT ON FUTURE TABLES IN SCHEMA analytics.raw TO ROLE loader;
-- ALL covers tables that already exist
-- FUTURE covers tables created later

GRANT OWNERSHIP ON TABLE analytics.raw.orders TO ROLE loader
  COPY CURRENT GRANTS;
-- the owner can alter and drop
-- COPY CURRENT GRANTS keeps the privileges already given out

CREATE FILE FORMAT analytics.raw.csv_ff
  TYPE = CSV
  FIELD_OPTIONALLY_ENCLOSED_BY = '"'
  SKIP_HEADER = 1
  NULL_IF = ('', 'NULL')
  EMPTY_FIELD_AS_NULL = TRUE;

CREATE STAGE analytics.raw.orders_stage
  FILE_FORMAT = analytics.raw.csv_ff;

PUT file://orders.csv @analytics.raw.orders_stage AUTO_COMPRESS = TRUE;

CREATE STAGE analytics.raw.orders_s3
  URL = 's3://example-bucket/orders/'
  STORAGE_INTEGRATION = orders_int
  FILE_FORMAT = analytics.raw.csv_ff;
-- PUT feeds an internal stage; an external stage points at cloud files

COPY INTO analytics.raw.orders (order_id, customer_id, amount, ordered_at)
FROM (
  SELECT $1, $2, $3, $4::TIMESTAMP_NTZ
  FROM @analytics.raw.orders_stage
)
FILE_FORMAT = (FORMAT_NAME = analytics.raw.csv_ff)
ON_ERROR = 'ABORT_STATEMENT';
-- $1 is the first file column; SKIP_HEADER already dropped the header
-- ON_ERROR = CONTINUE keeps the good rows and records the rest

COPY INTO analytics.raw.orders (order_id, customer_id, amount, ordered_at)
FROM (
  SELECT $1, $2, $3, $4::TIMESTAMP_NTZ
  FROM @analytics.raw.orders_stage
)
FILE_FORMAT = (FORMAT_NAME = analytics.raw.csv_ff)
VALIDATION_MODE = 'RETURN_ERRORS';

SELECT file_name, status, row_count, row_parsed, first_error_message
FROM information_schema.load_history
WHERE table_name = 'ORDERS'
ORDER BY last_load_time DESC;
-- VALIDATION_MODE loads nothing; it only returns errors

CREATE FILE FORMAT analytics.raw.json_ff TYPE = JSON;
CREATE TABLE analytics.raw.events (raw VARIANT);

COPY INTO analytics.raw.events (raw)
FROM (SELECT $1 FROM @analytics.raw.orders_stage)
FILE_FORMAT = (FORMAT_NAME = analytics.raw.json_ff);

SELECT
  raw:order_id::NUMBER AS order_id,
  raw:customer.name::STRING AS customer_name,
  raw:items[0].sku::STRING AS first_sku
FROM analytics.raw.events;
-- a missing path is NULL, not an error

SELECT
  e.raw:order_id::NUMBER AS order_id,
  item.value:sku::STRING AS sku,
  item.value:qty::NUMBER AS qty
FROM analytics.raw.events e,
LATERAL FLATTEN(input => e.raw:items) item;

MERGE INTO analytics.gold.orders t
USING analytics.silver.orders_staged s
  ON t.order_id = s.order_id
WHEN MATCHED AND s.updated_at >= t.updated_at THEN UPDATE SET
  amount = s.amount,
  updated_at = s.updated_at
WHEN NOT MATCHED THEN INSERT (order_id, customer_id, amount, ordered_at, updated_at)
  VALUES (s.order_id, s.customer_id, s.amount, s.ordered_at, s.updated_at);
-- running the same batch again should not insert a second row

SELECT *
FROM analytics.silver.orders_staged
QUALIFY ROW_NUMBER() OVER (
  PARTITION BY order_id ORDER BY updated_at DESC
) = 1;
-- use this as the MERGE source when one batch repeats a key

SELECT *
FROM analytics.raw.orders
WHERE updated_at > (
  SELECT COALESCE(MAX(updated_at), '1970-01-01'::TIMESTAMP_NTZ)
  FROM analytics.gold.orders
);
-- a row older than the watermark is skipped
-- a stream on the raw table still sees it

SELECT * FROM analytics.gold.orders AT (OFFSET => -900);
SELECT * FROM analytics.gold.orders
  AT (TIMESTAMP => DATEADD(minute, -30, CURRENT_TIMESTAMP()));
SELECT * FROM analytics.gold.orders BEFORE (STATEMENT => '<query_id>');
-- OFFSET is seconds ago; paste a real query id into BEFORE

CREATE TABLE analytics.gold.orders_restore
  CLONE analytics.gold.orders AT (OFFSET => -900);

ALTER TABLE analytics.gold.orders SWAP WITH analytics.gold.orders_restore;

UNDROP TABLE analytics.gold.orders;
-- UNDROP works only inside the Time Travel window
-- Fail-safe is a further 7 days on permanent tables, and you cannot query it

ALTER TABLE analytics.gold.orders SET DATA_RETENTION_TIME_IN_DAYS = 1;

CREATE TRANSIENT TABLE analytics.raw.orders_landing (
  order_id NUMBER,
  raw VARIANT
);
-- transient tables skip Fail-safe
-- a longer retention keeps more history and more storage

CREATE DATABASE analytics_dev CLONE analytics;
-- storage is shared until a row changes
-- grants on the database itself are not copied; check access before use

CREATE OR REPLACE STREAM analytics.raw.orders_stream
  ON TABLE analytics.raw.orders
  APPEND_ONLY = TRUE;

SELECT order_id, customer_id, amount, ordered_at, updated_at
FROM analytics.raw.orders_stream;
-- rows since the stream was last consumed by DML
-- APPEND_ONLY is inserts; a standard stream also shows updates and deletes

CREATE OR REPLACE TASK analytics.raw.merge_orders
  WAREHOUSE = analytics_wh
  SCHEDULE = '5 MINUTE'
  WHEN SYSTEM$STREAM_HAS_DATA('analytics.raw.orders_stream')
AS
  MERGE INTO analytics.gold.orders t
  USING (
    SELECT order_id, customer_id, amount, ordered_at, updated_at
    FROM analytics.raw.orders_stream
  ) s
    ON t.order_id = s.order_id
  WHEN MATCHED THEN UPDATE SET
    amount = s.amount,
    updated_at = s.updated_at
  WHEN NOT MATCHED THEN INSERT (order_id, customer_id, amount, ordered_at, updated_at)
    VALUES (s.order_id, s.customer_id, s.amount, s.ordered_at, s.updated_at);
-- a successful DML that reads the stream advances its offset
-- WHEN skips the warehouse when the stream is empty

CREATE OR REPLACE TASK analytics.gold.build_daily
  WAREHOUSE = analytics_wh
  AFTER analytics.raw.merge_orders
AS
  CREATE OR REPLACE TABLE analytics.gold.daily_sales AS
  SELECT ordered_at::DATE AS order_date, SUM(amount) AS revenue
  FROM analytics.gold.orders
  GROUP BY 1;

ALTER TASK analytics.gold.build_daily RESUME;
ALTER TASK analytics.raw.merge_orders RESUME;
-- the child has AFTER, not its own schedule
-- resume the child, then the root

CREATE OR REPLACE TASK analytics.raw.merge_orders_serverless
  USER_TASK_MANAGED_INITIAL_WAREHOUSE_SIZE = 'XSMALL'
  SCHEDULE = '5 MINUTE'
  WHEN SYSTEM$STREAM_HAS_DATA('analytics.raw.orders_stream')
AS
  MERGE INTO analytics.gold.orders t
  USING (
    SELECT order_id, customer_id, amount, ordered_at, updated_at
    FROM analytics.raw.orders_stream
  ) s
    ON t.order_id = s.order_id
  WHEN NOT MATCHED THEN INSERT (order_id, customer_id, amount, ordered_at, updated_at)
    VALUES (s.order_id, s.customer_id, s.amount, s.ordered_at, s.updated_at);
-- no WAREHOUSE clause; Snowflake sizes the compute
-- an append-only stream is inserts, so NOT MATCHED is the path that fires

CREATE OR REPLACE PIPE analytics.raw.orders_pipe
  AUTO_INGEST = TRUE
AS
  COPY INTO analytics.raw.orders (order_id, customer_id, amount, ordered_at)
  FROM (
    SELECT $1, $2, $3, $4::TIMESTAMP_NTZ
    FROM @analytics.raw.orders_s3
  )
  FILE_FORMAT = (FORMAT_NAME = analytics.raw.csv_ff);
-- a cloud notification wakes the pipe for new files
-- a large nightly drop is often a scheduled COPY or a task instead

ALTER PIPE analytics.raw.orders_pipe REFRESH;

SELECT SYSTEM$PIPE_STATUS('analytics.raw.orders_pipe');
-- REFRESH copies files already sitting in the stage
-- read pendingFileCount and the last error before calling the pipe healthy

CREATE OR REPLACE DYNAMIC TABLE analytics.gold.daily_orders
  TARGET_LAG = '15 minutes'
  WAREHOUSE = analytics_wh
  REFRESH_MODE = AUTO
AS
  SELECT ordered_at::DATE AS order_date, SUM(amount) AS revenue
  FROM analytics.gold.orders
  GROUP BY 1;
-- you declare the SELECT and a lag; Snowflake refreshes it
-- AUTO, INCREMENTAL, or FULL; a task graph is the manual version

ALTER WAREHOUSE analytics_wh SET
  AUTO_SUSPEND = 60
  STATEMENT_TIMEOUT_IN_SECONDS = 1800;

SELECT warehouse_name, SUM(credits_used) AS credits
FROM snowflake.account_usage.warehouse_metering_history
WHERE start_time > DATEADD(day, -7, CURRENT_TIMESTAMP())
GROUP BY 1
ORDER BY 2 DESC;
-- account_usage can lag by about an hour

CREATE RESOURCE MONITOR analytics_budget
  WITH CREDIT_QUOTA = 100
  FREQUENCY = MONTHLY
  START_TIMESTAMP = IMMEDIATELY
  TRIGGERS
    ON 80 PERCENT DO NOTIFY
    ON 100 PERCENT DO SUSPEND
    ON 110 PERCENT DO SUSPEND_IMMEDIATE;

ALTER WAREHOUSE analytics_wh SET RESOURCE_MONITOR = analytics_budget;

CREATE OR REPLACE MASKING POLICY analytics.gold.mask_email AS (val STRING)
  RETURNS STRING ->
    CASE
      WHEN IS_ROLE_IN_SESSION('PII_ADMIN') THEN val
      ELSE '***'
    END;

ALTER TABLE analytics.gold.customers MODIFY COLUMN email
  SET MASKING POLICY analytics.gold.mask_email;

CREATE TAG analytics.gold.pii ALLOWED_VALUES 'email', 'name';
ALTER TABLE analytics.gold.customers MODIFY COLUMN email
  SET TAG analytics.gold.pii = 'email';

CREATE OR REPLACE ROW ACCESS POLICY analytics.gold.region_policy
  AS (region STRING) RETURNS BOOLEAN ->
    IS_ROLE_IN_SESSION('ANALYST_ALL') OR region = 'IN';

ALTER TABLE analytics.gold.customers
  ADD ROW ACCESS POLICY analytics.gold.region_policy ON (region);

SELECT query_id, query_start_time
FROM snowflake.account_usage.access_history
WHERE query_start_time > DATEADD(day, -1, CURRENT_TIMESTAMP());
-- a masking policy hides a value; a row policy hides rows