Getting started with Snowflake: warehouses, loading data and time travel

CREATE DATABASE IF NOT EXISTS analytics;
CREATE SCHEMA IF NOT EXISTS analytics.raw;       -- data as it arrives
CREATE SCHEMA IF NOT EXISTS analytics.staging;   -- cleaned and renamed
CREATE SCHEMA IF NOT EXISTS analytics.marts;     -- facts and dimensions for reporting

USE DATABASE analytics;
USE SCHEMA raw;

CREATE WAREHOUSE IF NOT EXISTS transform_wh
    WAREHOUSE_SIZE = 'XSMALL'
    AUTO_SUSPEND = 60          -- seconds of inactivity before pausing
    AUTO_RESUME = TRUE
    INITIALLY_SUSPENDED = TRUE;

USE WAREHOUSE transform_wh;

CREATE OR REPLACE FILE FORMAT csv_format
    TYPE = CSV
    SKIP_HEADER = 1
    FIELD_OPTIONALLY_ENCLOSED_BY = '"'
    NULL_IF = ('', 'NULL');

CREATE OR REPLACE STAGE raw_stage FILE_FORMAT = csv_format;

-- From SnowSQL or the Snowflake CLI, upload a local file:
-- PUT file://orders_2026_09.csv @raw_stage;

CREATE OR REPLACE TABLE raw.orders (
    order_id     NUMBER,
    customer_id  NUMBER,
    order_date   DATE,
    amount       NUMBER(10, 2)
);

COPY INTO raw.orders
FROM @raw_stage
PATTERN = '.*orders_.*[.]csv'
ON_ERROR = 'ABORT_STATEMENT';

CREATE OR REPLACE TABLE raw.events (payload VARIANT);

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

-- The table as it was 30 minutes ago
SELECT * FROM marts.fct_sales AT(OFFSET => -60 * 30);

-- Recover from an accidental delete by re-inserting yesterday's rows
INSERT INTO marts.fct_sales
SELECT * FROM marts.fct_sales AT(TIMESTAMP => DATEADD(day, -1, CURRENT_TIMESTAMP()))
WHERE order_date = '2026-09-29';

-- Bring back a dropped table
UNDROP TABLE marts.fct_sales;

CREATE DATABASE analytics_dev CLONE analytics;
CREATE TABLE marts.fct_sales_backup CLONE marts.fct_sales;

CREATE OR REPLACE STREAM raw.orders_stream ON TABLE raw.orders;

CREATE OR REPLACE TASK load_orders
    WAREHOUSE = transform_wh
    SCHEDULE = 'USING CRON 0 6 * * * Europe/London'
WHEN SYSTEM$STREAM_HAS_DATA('raw.orders_stream')
AS
INSERT INTO staging.orders
SELECT order_id, customer_id, order_date, amount
FROM raw.orders_stream
WHERE METADATA$ACTION = 'INSERT';

ALTER TASK load_orders RESUME;   -- tasks are created suspended