-- Example 1: Create standard stream
-- SHOW_INITIAL_ROWS=TRUE captures existing data as INSERT events
-- This is useful for backfilling downstream tables
CREATE OR REPLACE STREAM sales_stream
ON TABLE sales_transactions
SHOW_INITIAL_ROWS = TRUE    -- Include current rows as initial INSERT events
COMMENT = 'CDC stream for sales transactions';

-- Example 2: Create append-only stream
-- APPEND_ONLY=TRUE reduces overhead by tracking only inserts
-- Cannot detect UPDATE or DELETE operations
CREATE OR REPLACE STREAM immutable_log_stream
ON TABLE audit_log
APPEND_ONLY = TRUE           -- Only track INSERT operations
COMMENT = 'Append-only stream for audit logs';

-- Example 3: Query stream with change metadata
-- METADATA$ACTION: The DML operation (INSERT, UPDATE, DELETE)
-- METADATA$ISUPDATE: TRUE if the row was updated (not a new insert)
-- METADATA$ROW_ID: Unique identifier for this change event
SELECT 
    *,
    METADATA$ACTION as change_action,
    METADATA$ISUPDATE as is_update,
    METADATA$ROW_ID as row_id
FROM sales_stream;

-- Example 4: Process stream with merge
-- MERGE handles INSERT, UPDATE, and DELETE in a single atomic operation
-- METADATA$ACTION determines which WHEN clause to execute
MERGE INTO sales_target t
USING sales_stream s
ON t.id = s.id
WHEN MATCHED AND s.METADATA$ACTION = 'DELETE' THEN
    DELETE
WHEN MATCHED AND s.METADATA$ACTION = 'UPDATE' THEN
    UPDATE SET 
        t.product = s.product,
        t.amount = s.amount,
        t.updated_at = s.updated_at
WHEN NOT MATCHED AND s.METADATA$ACTION = 'INSERT' THEN
    INSERT (id, product, amount, updated_at)
    VALUES (s.id, s.product, s.amount, s.updated_at);

-- Example 5: Stream offset management
-- Check current stream offset (how many unconsumed changes exist)
SELECT SYSTEM$STREAM_HAS_DATA('sales_stream') AS has_data;

-- Check stream offset details
SELECT 
    'sales_stream' AS stream_name,
    SYSTEM$STREAM_GET_TABLE_TIMESTAMP('sales_stream') AS table_timestamp,
    SYSTEM$STREAM_HAS_DATA('sales_stream') AS has_pending_changes;

-- Example 6: Create insert-only stream for append-only source tables
-- INSERT_ONLY is the most lightweight stream type
-- Best for tables that only receive INSERT operations (e.g., logs, events)
CREATE OR REPLACE STREAM event_log_stream
ON TABLE event_log
INSERT_ONLY = TRUE
COMMENT = 'Lightweight stream for append-only event logs';

-- Example 7: Stream on external tables for data lake CDC
-- Streams can track changes on external tables (Parquet, JSON, etc.)
CREATE OR REPLACE STREAM external_data_stream
ON TABLE external_orders
APPEND_ONLY = TRUE
COMMENT = 'CDC stream for external table data lake ingestion';

-- Example 8: Create basic task
-- CRON syntax: minute hour day-of-month month day-of-week
-- '0 3 * * *' means daily at 3:00 AM
-- America/New_York timezone ensures correct local execution
CREATE OR REPLACE TASK daily_cleanup_task
    WAREHOUSE = 'maintenance_wh'                              -- Dedicated warehouse for task
    SCHEDULE = 'USING CRON 0 3 * * * America/New_York'       -- Daily at 3 AM EST
    COMMENT = 'Daily cleanup of old data'
AS
    DELETE FROM audit_log 
    WHERE created_at < DATEADD(day, -90, CURRENT_TIMESTAMP());

-- Example 9: Create task with error handling
-- ERROR_INTEGRATION specifies notification service for failures
-- ALLOW_OVERLAPPING_EXECUTION=FALSE prevents concurrent runs
-- BEGIN...EXCEPTION block provides structured error handling
CREATE OR REPLACE TASK etl_task
    WAREHOUSE = 'etl_wh'
    SCHEDULE = 'USING CRON 0 2 * * * America/New_York'
    ERROR_INTEGRATION = 's3_error_notifications'              -- Notify on failure
    ALLOW_OVERLAPPING_EXECUTION = FALSE                       -- No concurrent executions
    USER_TASK_TIMEOUT_MS = 7200000                            -- 2-hour timeout
    SUSPEND_TASK_AFTER_NUM_FAILURES = 5                       -- Auto-suspend after 5 failures
AS
BEGIN
    -- Log start
    INSERT INTO task_log (task_name, start_time, status)
    VALUES ('etl_task', CURRENT_TIMESTAMP(), 'RUNNING');
    
    -- Execute ETL
    INSERT INTO target_table 
    SELECT * FROM source_table 
    WHERE date = CURRENT_DATE();
    
    -- Log success
    INSERT INTO task_log (task_name, start_time, status, end_time)
    VALUES ('etl_task', CURRENT_TIMESTAMP(), 'SUCCESS', CURRENT_TIMESTAMP());
    
    EXCEPTION
        WHEN OTHER THEN
            -- Log failure with error details
            INSERT INTO task_log (task_name, start_time, status, error_message)
            VALUES ('etl_task', CURRENT_TIMESTAMP(), 'FAILED', SQLERRM);
            RAISE;  -- Re-raise to mark task as FAILED
END;

-- Example 10: Interval-based scheduling (alternative to CRON)
-- Runs every 15 minutes using INTERVAL syntax
CREATE OR REPLACE TASK frequent_sync_task
    WAREHOUSE = 'sync_wh'
    SCHEDULE = '60 * * * *'          -- Every 60 minutes
AS
    INSERT INTO staging_data 
    SELECT * FROM source_table 
    WHERE updated_at > DATEADD(minute, -60, CURRENT_TIMESTAMP());

-- Example 11: Create task graph (DAG)
-- Root task runs on schedule; child tasks run after parent succeeds
-- Parallel branches execute independently; final_task waits for all

CREATE OR REPLACE TASK root_task
    WAREHOUSE = 'etl_wh'
    SCHEDULE = 'USING CRON 0 2 * * * America/New_York'
AS
    INSERT INTO raw_sales SELECT * FROM stage_sales;

CREATE OR REPLACE TASK child_task_1
    WAREHOUSE = 'etl_wh'
    AFTER root_task                  -- Runs after root_task completes successfully
AS
    INSERT INTO clean_sales SELECT * FROM raw_sales WHERE amount > 0;

CREATE OR REPLACE TASK child_task_2
    WAREHOUSE = 'etl_wh'
    AFTER root_task                  -- Parallel branch: also runs after root_task
AS
    INSERT INTO raw_inventory SELECT * FROM stage_inventory;

CREATE OR REPLACE TASK final_task
    WAREHOUSE = 'etl_wh'
    AFTER child_task_1, child_task_2  -- Synchronization point: waits for BOTH
AS
    INSERT INTO analytics_sales 
    SELECT s.*, i.stock 
    FROM clean_sales s 
    JOIN raw_inventory i ON s.product_id = i.product_id;

-- Example 12: Resume/suspend task graph
-- Must resume child tasks before parent tasks
ALTER TASK child_task_1 RESUME;
ALTER TASK child_task_2 RESUME;
ALTER TASK root_task RESUME;

-- Suspend in reverse order (parent first)
ALTER TASK root_task SUSPEND;
ALTER TASK child_task_1 SUSPEND;
ALTER TASK child_task_2 SUSPEND;

-- Example 13: Alter task schedule
ALTER TASK daily_cleanup_task RESUME;
ALTER TASK daily_cleanup_task SUSPEND;
ALTER TASK daily_cleanup_task SET SCHEDULE = 'USING CRON 0 4 * * * America/New_York';

-- Example 14: Monitor task execution history
-- TASK_HISTORY provides detailed execution metrics per task run
SELECT 
    task_name,
    query_id,
    state,                                          -- SUCCEEDED, FAILED, BLOCKED, EXECUTING
    scheduled_time,                                 -- When task was scheduled to run
    started_time,                                   -- When task actually started executing
    completed_time,                                 -- When task finished
    DATEDIFF(second, started_time, completed_time) AS duration_seconds,
    error_code,
    error_message
FROM TABLE(INFORMATION_SCHEMA.TASK_HISTORY(
    SCHEDULED_TIME_RANGE_START => DATEADD(day, -7, CURRENT_TIMESTAMP())
))
ORDER BY scheduled_time DESC;

-- Example 15: Monitor task graph dependencies
-- Identify blocked tasks and their parent dependencies
SELECT 
    t.task_name,
    t.state AS task_state,
    d.dependency_type,
    d.downstream_task_name AS parent_task,
    p.state AS parent_state
FROM INFORMATION_SCHEMA.TASK_DEPENDENCIES d
JOIN INFORMATION_SCHEMA.TASKS t ON d.task_name = t.task_name
JOIN INFORMATION_SCHEMA.TASKS p ON d.downstream_task_name = p.task_name
WHERE t.state = 'BLOCKED'
ORDER BY t.task_name;

-- Example 16: Stream consumption monitoring
-- Track stream lag to prevent unbounded growth
SELECT 
    stream_name,
    table_name,
    SYSTEM$STREAM_HAS_DATA(stream_name) AS has_pending_data,
    created_on,
    owner_name
FROM INFORMATION_SCHEMA.STREAMS
ORDER BY created_on DESC;

-- Example 17: Stream consumption pattern with error handling
CREATE OR REPLACE PROCEDURE process_stream_changes()
RETURNS VARCHAR
LANGUAGE SQL
AS
$$
DECLARE
    change_count INTEGER;
BEGIN
    -- Count available changes in the stream
    SELECT COUNT(*) INTO change_count FROM sales_stream;
    
    -- Process only if changes exist
    IF (change_count > 0) THEN
        -- Process INSERT operations
        INSERT INTO sales_archive (id, product, amount, updated_at)
        SELECT id, product, amount, updated_at
        FROM sales_stream
        WHERE METADATA$ACTION = 'INSERT';
        
        -- Process UPDATE operations
        UPDATE sales_archive t
        SET 
            product = s.product,
            amount = s.amount,
            updated_at = s.updated_at
        FROM sales_stream s
        WHERE t.id = s.id 
          AND s.METADATA$ACTION = 'UPDATE'
          AND s.METADATA$ISUPDATE = TRUE;
        
        -- Process DELETE operations
        DELETE FROM sales_archive t
        WHERE EXISTS (
            SELECT 1 FROM sales_stream s
            WHERE t.id = s.id 
              AND s.METADATA$ACTION = 'DELETE'
        );
        
        RETURN 'Processed ' || change_count || ' changes';
    ELSE
        RETURN 'No changes to process';
    END IF;
END;
$$;

-- Example 18: Task graph with conditional execution
CREATE OR REPLACE TASK conditional_etl_task
    WAREHOUSE = 'etl_wh'
    SCHEDULE = 'USING CRON 0 2 * * * America/New_York'
AS
BEGIN
    -- Only process if source data exists
    IF (SELECT COUNT(*) FROM source_table WHERE load_date = CURRENT_DATE()) > 0 THEN
        INSERT INTO target_table 
        SELECT * FROM source_table 
        WHERE load_date = CURRENT_DATE();
    END IF;
END;

-- Example 19: Multi-cluster warehouse for task execution
-- Ensures sufficient concurrency for parallel task graphs
CREATE OR REPLACE TASK high_volume_etl
    WAREHOUSE = 'etl_wh_scaled'
    SCHEDULE = 'USING CRON 0 1 * * * America/New_York'
AS
    INSERT INTO aggregated_data 
    SELECT region, SUM(amount) 
    FROM raw_sales 
    GROUP BY region;

-- Configure warehouse for auto-scaling
ALTER WAREHOUSE etl_wh_scaled 
    SET 
        MIN_CLUSTER_COUNT = 1,
        MAX_CLUSTER_COUNT = 5,
        SCALING_POLICY = 'ECONOMY';

-- Example 20: Drop stream (requires no pending changes or explicitly drop)
-- Streams with unconsumed data can be dropped (offset is lost)
DROP STREAM IF EXISTS old_stream CASCADE;