-- Basic COPY INTO from external stage (Azure ADLS)
COPY INTO raw_bookings
FROM @azure_booking_stage/2026/03/25/        -- Read from this ADLS path
FILE_FORMAT = (
    TYPE = 'PARQUET'                          -- File type
)
ON_ERROR = 'CONTINUE';                        -- Skip bad files, continue loading
-- Options: ABORT_STATEMENT (default), CONTINUE, SKIP_FILE

-- COPY INTO with transformation (load + clean in one step)
COPY INTO silver_bookings (booking_id, booking_date, amount, airline_code)
FROM (
    SELECT
        $1:booking_id::VARCHAR,               -- Cast from VARIANT
        $1:booking_date::DATE,
        $1:fare.amount::NUMBER(10,2),
        UPPER($1:airline_code::VARCHAR)        -- Transform during load
    FROM @azure_booking_stage/bookings.json
)
FILE_FORMAT = (TYPE = 'JSON');

-- COPY INTO with error handling — see what failed
COPY INTO raw_bookings FROM @stage/
ON_ERROR = 'CONTINUE';

-- After loading, check what errors occurred:
SELECT * FROM TABLE(VALIDATE(raw_bookings, JOB_ID => '_last'));
-- Shows exactly which files + rows failed and why

-- COPY INTO is idempotent (safe to re-run!)
-- Snowflake tracks which files were already loaded (load history table)
-- Re-running the same COPY INTO skips already-loaded files automatically
-- To force reload: COPY INTO ... FORCE = TRUE;

-- Create a pipe that auto-loads from a stage
CREATE PIPE booking_pipe
    AUTO_INGEST = TRUE          -- Enabled for cloud event notifications
    AS
    COPY INTO raw_bookings      -- What to run when files arrive
    FROM @azure_booking_stage
    FILE_FORMAT = (TYPE = 'PARQUET');

-- Check pipe status
SELECT SYSTEM$PIPE_STATUS('booking_pipe');
-- Shows: pendingFileCount, lastIngestedTimestamp, status

-- See load history (what files loaded, errors)
SELECT * FROM TABLE(INFORMATION_SCHEMA.COPY_HISTORY(
    TABLE_NAME => 'raw_bookings',
    START_TIME => DATEADD('hour', -24, CURRENT_TIMESTAMP)
));

-- ========================================
-- CREATING STREAMS
-- ========================================

-- Standard stream (tracks all DML changes)
CREATE STREAM booking_changes_stream
    ON TABLE raw_bookings
    SHOW_INITIAL_ROWS = FALSE;   -- Only track NEW changes from this point
    -- SHOW_INITIAL_ROWS = TRUE → backfill: include all existing rows as INSERTs

-- Append-only stream (lighter weight — only new inserts)
CREATE STREAM new_bookings_stream
    ON TABLE raw_bookings
    APPEND_ONLY = TRUE;
-- Use for event/log tables where rows are never updated or deleted

-- Insert-only stream on an external table
CREATE STREAM new_files_stream
    ON EXTERNAL TABLE ext_booking_files
    INSERT_ONLY = TRUE;
-- Fires when new files appear in the external stage

-- ========================================
-- QUERYING A STREAM (peek at changes)
-- ========================================

SELECT *,
    METADATA$ACTION,         -- 'INSERT' or 'DELETE'
    METADATA$ISUPDATE,       -- TRUE if part of an UPDATE
    METADATA$ROW_ID          -- Internal row identifier
FROM booking_changes_stream;
-- This is a READ — it does NOT consume the stream
-- Stream is only consumed when changes are used in a DML statement

-- ========================================
-- CONSUMING A STREAM (the MERGE pattern — #1 interview pattern!)
-- ========================================

-- The Stream + MERGE pattern: upsert changes into a target table
MERGE INTO silver_bookings AS target
USING (
    SELECT booking_id, booking_date, amount, airline_code,
           METADATA$ACTION AS action,
           METADATA$ISUPDATE AS is_update
    FROM booking_changes_stream
) AS src
ON target.booking_id = src.booking_id
-- Handle inserts (new rows)
WHEN NOT MATCHED AND src.action = 'INSERT'
    THEN INSERT (booking_id, booking_date, amount, airline_code)
         VALUES (src.booking_id, src.booking_date, src.amount, src.airline_code)
-- Handle updates (new value of updated rows)
WHEN MATCHED AND src.action = 'INSERT' AND src.is_update = TRUE
    THEN UPDATE SET
        target.amount = src.amount,
        target.booking_date = src.booking_date,
        target.airline_code = src.airline_code
-- Handle deletes
WHEN MATCHED AND src.action = 'DELETE' AND src.is_update = FALSE
    THEN DELETE;

-- After MERGE commits → stream advances → consumed rows disappear
-- Next query to stream shows only changes AFTER this MERGE

-- ========================================
-- STALE STREAMS — how to avoid them
-- ========================================

-- A stream becomes STALE if you don't consume it within the table's
-- DATA_RETENTION_TIME_IN_DAYS (Time Travel period, default 1 day).
-- Once stale, the stream is BROKEN — you must recreate it.

-- Check if a stream is stale:
SHOW STREAMS LIKE 'booking_changes_stream';
-- Look at "stale_after" column — if past, the stream is stale

-- Prevention: consume streams regularly OR extend retention:
ALTER TABLE raw_bookings SET DATA_RETENTION_TIME_IN_DAYS = 14;
-- Now stream has 14 days before going stale
-- Trade-off: longer retention = more storage cost (Time Travel storage)

-- ========================================
-- CREATING TASKS
-- ========================================

-- Simple task: run SQL every 5 minutes using a warehouse
CREATE TASK load_bronze_task
    WAREHOUSE = 'ETL_WH'               -- Uses this warehouse's credits
    SCHEDULE = '5 MINUTE'              -- Interval schedule
AS
    COPY INTO raw_bookings
    FROM @azure_booking_stage;

-- Cron-based schedule
CREATE TASK nightly_aggregate_task
    WAREHOUSE = 'ETL_WH'
    SCHEDULE = 'USING CRON 0 2 * * * UTC'   -- 2:00 AM UTC daily
    -- Cron: minute hour day month weekday timezone
AS
    CALL refresh_daily_aggregates();

-- !! CRITICAL: Tasks are SUSPENDED by default !!
-- You MUST resume them or they will never run
ALTER TASK load_bronze_task RESUME;
ALTER TASK load_bronze_task SUSPEND;     -- Pause it

-- ========================================
-- SERVERLESS TASKS vs WAREHOUSE-BASED TASKS
-- ========================================

-- Warehouse-based: uses a specific warehouse (you control size & cost)
CREATE TASK warehouse_task
    WAREHOUSE = 'ETL_WH'
    SCHEDULE = '10 MINUTE'
AS SELECT 1;

-- Serverless: Snowflake manages compute — no warehouse needed
CREATE TASK serverless_task
    USER_TASK_MANAGED_INITIAL_WAREHOUSE_SIZE = 'MEDIUM'
    -- Omit WAREHOUSE = ... to make it serverless
    SCHEDULE = '5 MINUTE'
AS
    INSERT INTO audit_log VALUES (CURRENT_TIMESTAMP, 'heartbeat');

-- Serverless tasks:
-- ✓ ~10% cheaper than warehouse-based for equivalent work
-- ✓ No warehouse to manage — auto-scales to workload
-- ✓ Best for lightweight, frequent tasks
-- ✗ Less control over compute resources
-- ✗ Compute size is approximate (Snowflake decides)

-- ========================================
-- TASK TREES (DAG — parent → child dependencies)
-- ========================================

-- Root task: has a SCHEDULE (the only task that needs one)
CREATE TASK bronze_task
    WAREHOUSE = 'ETL_WH'
    SCHEDULE = 'USING CRON 0 2 * * * UTC'  -- 2 AM daily
AS
    CALL load_bronze_data();

-- Child task: uses AFTER clause (no SCHEDULE — triggered by parent)
CREATE TASK silver_task
    WAREHOUSE = 'ETL_WH'
    AFTER bronze_task               -- Runs ONLY after bronze_task succeeds
AS
    CALL transform_silver_data();

-- Grandchild task: runs after silver
CREATE TASK gold_task
    WAREHOUSE = 'ETL_WH'
    AFTER silver_task
AS
    CALL aggregate_gold_metrics();

-- A child can depend on MULTIPLE parents (fan-in):
CREATE TASK final_report_task
    WAREHOUSE = 'ETL_WH'
    AFTER silver_task, gold_task     -- Waits for BOTH to finish
AS
    CALL generate_report();

-- !! Resume ONLY the root task — children auto-activate !!
-- !! But you must RESUME children BEFORE resuming the root !!
ALTER TASK gold_task RESUME;
ALTER TASK silver_task RESUME;
ALTER TASK bronze_task RESUME;      -- Root task last

-- ========================================
-- CONDITIONAL EXECUTION: SYSTEM$STREAM_HAS_DATA()
-- ========================================

-- Only run the task if the stream has new data (skip empty cycles)
CREATE TASK process_booking_changes
    WAREHOUSE = 'ETL_WH'
    SCHEDULE = '5 MINUTE'
    WHEN SYSTEM$STREAM_HAS_DATA('booking_changes_stream')
    -- If stream is empty → task is SKIPPED (no warehouse credit used!)
AS
    MERGE INTO silver_bookings AS target
    USING booking_changes_stream AS src
    ON target.booking_id = src.booking_id
    WHEN NOT MATCHED AND src.METADATA$ACTION = 'INSERT'
        THEN INSERT (booking_id, booking_date, amount, airline_code)
             VALUES (src.booking_id, src.booking_date, src.amount, src.airline_code)
    WHEN MATCHED AND src.METADATA$ACTION = 'INSERT' AND src.METADATA$ISUPDATE = TRUE
        THEN UPDATE SET target.amount = src.amount
    WHEN MATCHED AND src.METADATA$ACTION = 'DELETE' AND src.METADATA$ISUPDATE = FALSE
        THEN DELETE;

ALTER TASK process_booking_changes RESUME;

-- ========================================
-- COMPLETE STREAM + TASK PIPELINE (end-to-end example)
-- ========================================

-- Step 1: Source table exists
-- CREATE TABLE raw_bookings (booking_id VARCHAR, booking_date DATE, ...);

-- Step 2: Create stream to track changes
CREATE STREAM booking_cdc_stream ON TABLE raw_bookings;

-- Step 3: Create task that consumes stream via MERGE
CREATE TASK cdc_pipeline_task
    WAREHOUSE = 'ETL_WH'
    SCHEDULE = '5 MINUTE'
    WHEN SYSTEM$STREAM_HAS_DATA('booking_cdc_stream')
AS
    MERGE INTO silver_bookings AS t
    USING booking_cdc_stream AS s
    ON t.booking_id = s.booking_id
    WHEN MATCHED AND s.METADATA$ACTION = 'DELETE' AND s.METADATA$ISUPDATE = FALSE
        THEN DELETE
    WHEN MATCHED AND s.METADATA$ACTION = 'INSERT' AND s.METADATA$ISUPDATE = TRUE
        THEN UPDATE SET t.amount = s.amount, t.booking_date = s.booking_date
    WHEN NOT MATCHED AND s.METADATA$ACTION = 'INSERT'
        THEN INSERT (booking_id, booking_date, amount)
             VALUES (s.booking_id, s.booking_date, s.amount);

-- Step 4: Resume the task
ALTER TASK cdc_pipeline_task RESUME;

-- Now: any INSERT/UPDATE/DELETE on raw_bookings is automatically
-- captured by the stream and merged into silver_bookings every 5 minutes

-- ========================================
-- MONITORING TASKS
-- ========================================

-- View task run history (recent runs, status, errors)
SELECT name, state, scheduled_time, completed_time,
       error_code, error_message,
       DATEDIFF('second', query_start_time, completed_time) AS duration_sec
FROM TABLE(INFORMATION_SCHEMA.TASK_HISTORY(
    TASK_NAME => 'CDC_PIPELINE_TASK',
    SCHEDULED_TIME_RANGE_START => DATEADD('day', -1, CURRENT_TIMESTAMP)
))
ORDER BY scheduled_time DESC;
-- state: 'SUCCEEDED', 'FAILED', 'SKIPPED' (WHEN condition was false)

-- Account-level task history (all tasks)
SELECT * FROM SNOWFLAKE.ACCOUNT_USAGE.TASK_HISTORY
WHERE scheduled_time > DATEADD('hour', -24, CURRENT_TIMESTAMP)
ORDER BY scheduled_time DESC;

-- ========================================
-- CREATING DYNAMIC TABLES
-- ========================================

-- Silver layer: clean + transform raw data
CREATE DYNAMIC TABLE silver_bookings
    TARGET_LAG = '5 minutes'         -- Freshness guarantee
    WAREHOUSE = 'ETL_WH'             -- Warehouse for refresh compute
AS
SELECT
    booking_id,
    booking_date,
    UPPER(airline_code)     AS airline_code,
    amount,
    CURRENT_TIMESTAMP       AS processed_at
FROM raw_bookings
WHERE amount > 0
  AND booking_id IS NOT NULL;

-- Snowflake AUTOMATICALLY:
-- 1. Detects when raw_bookings changes
-- 2. Computes ONLY the incremental diff (not full reprocess)
-- 3. Updates silver_bookings within 5 minutes of source change

-- Gold layer: aggregate (auto-chains to silver!)
CREATE DYNAMIC TABLE gold_daily_revenue
    TARGET_LAG = '1 hour'
    WAREHOUSE = 'ETL_WH'
AS
SELECT
    booking_date,
    airline_code,
    SUM(amount)  AS daily_revenue,
    COUNT(*)     AS booking_count
FROM silver_bookings    -- Reads from dynamic table above!
GROUP BY booking_date, airline_code;
-- Snowflake auto-detects the silver → gold dependency
-- No AFTER clause needed — it just works

-- Manually refresh (force update now, useful for testing)
ALTER DYNAMIC TABLE silver_bookings REFRESH;

-- Suspend / resume
ALTER DYNAMIC TABLE silver_bookings SUSPEND;
ALTER DYNAMIC TABLE silver_bookings RESUME;

-- Check status of all dynamic tables
SELECT name, schema_name, target_lag, refresh_mode,
       refresh_mode_reason, data_timestamp
FROM INFORMATION_SCHEMA.DYNAMIC_TABLES;
-- refresh_mode: 'INCREMENTAL' (optimal) or 'FULL' (fallback if query is too complex)
-- data_timestamp: when the data was last refreshed to

-- ========================================
-- CREATING A MATERIALIZED VIEW
-- ========================================

CREATE MATERIALIZED VIEW daily_revenue_mv AS
SELECT
    booking_date,
    airline_code,
    SUM(amount)   AS total_revenue,
    COUNT(*)      AS booking_count,
    AVG(amount)   AS avg_booking_value
FROM fact_bookings
GROUP BY booking_date, airline_code;
-- Snowflake physically stores this result
-- Auto-refreshes when fact_bookings changes

-- Query the MV (same as querying a table — but pre-computed!)
SELECT * FROM daily_revenue_mv
WHERE booking_date = '2026-03-25' AND airline_code = 'LH';
-- Returns instantly — reads pre-aggregated result, not the full fact table

-- Snowflake's optimizer can AUTOMATICALLY route queries to the MV
-- Even if you query the BASE TABLE, Snowflake may use the MV if it
-- detects the query matches the MV's definition (query rewrite)

-- Drop a materialized view
DROP MATERIALIZED VIEW daily_revenue_mv;

-- Check MV refresh status
SELECT * FROM SNOWFLAKE.ACCOUNT_USAGE.MATERIALIZED_VIEW_REFRESH_HISTORY
WHERE materialized_view_name = 'DAILY_REVENUE_MV'
ORDER BY start_time DESC;

-- STEP 1: Find slow queries
SELECT query_text, total_elapsed_time/1000 as seconds,
       bytes_scanned/1024/1024/1024 as gb_scanned,
       partitions_scanned, partitions_total,
       ROUND(partitions_scanned*100.0/partitions_total, 1) as pct_scanned
FROM SNOWFLAKE.ACCOUNT_USAGE.QUERY_HISTORY
WHERE execution_status = 'SUCCESS'
  AND warehouse_name = 'BI_WH'
ORDER BY total_elapsed_time DESC
LIMIT 20;
-- HIGH pct_scanned (>80%) = bad pruning → add clustering key
-- LOW gb_scanned but slow = compute bottleneck → scale up warehouse

-- STEP 2: Check clustering health
SELECT SYSTEM$CLUSTERING_INFORMATION('bookings', '(booking_date, airline_code)');
-- average_depth > 6 → reclustering needed
-- average_overlaps > 5 → partitions heavily mixed

-- STEP 3: Use Materialized Views for repeated heavy aggregations
CREATE MATERIALIZED VIEW daily_revenue_mv AS
SELECT booking_date, airline_code, SUM(amount) as revenue
FROM silver_bookings
GROUP BY booking_date, airline_code;
-- Snowflake pre-computes and auto-refreshes this view
-- BI queries that aggregate by date+airline now hit MV instead of raw table
-- 100x faster for common aggregation queries!

-- STEP 4: Search Optimization Service (for needle-in-haystack queries)
ALTER TABLE bookings ADD SEARCH OPTIMIZATION;
-- Optimizes: WHERE booking_id = 'BK123456' (point lookup)
-- Optimizes: WHERE passenger_email = 'user@example.com' (equality on high-cardinality col)
-- NOT for range queries (use clustering key instead)

-- STEP 5: Query Profile analysis
-- After running a query, in Snowflake UI: Query History → Click query → Query Profile
-- Shows execution tree, where time was spent
-- Key metrics to look for:
--   "Bytes scanned from cache" high = good (cache working)
--   "Partitions scanned" vs "Partitions total" → pruning efficiency
--   "Spilling to disk" = not enough memory → scale up warehouse