Tasks and Streams¶

-- CRON-based schedule
CREATE TASK hourly_load
  WAREHOUSE = pipeline_wh
  SCHEDULE = 'USING CRON 0 * * * * America/New_York'
AS
  COPY INTO raw_data FROM @source_stage;

-- Interval-based schedule (minutes)
CREATE TASK frequent_check
  WAREHOUSE = pipeline_wh
  SCHEDULE = '5 MINUTE'
AS
  CALL check_data_quality();

-- Root task
CREATE TASK ingest_task
  WAREHOUSE = pipeline_wh
  SCHEDULE = '10 MINUTE'
  WHEN SYSTEM$STREAM_HAS_DATA('source_stream')
AS
  INSERT INTO staging FROM source_stream;

-- First-level children (run after ingest_task)
CREATE TASK transform_a
  WAREHOUSE = pipeline_wh
  AFTER ingest_task
AS
  INSERT INTO dim_customers SELECT ... FROM staging;

CREATE TASK transform_b
  WAREHOUSE = pipeline_wh
  AFTER ingest_task
AS
  INSERT INTO dim_products SELECT ... FROM staging;

-- Second-level child (runs after BOTH transform_a and transform_b)
CREATE TASK aggregate_task
  WAREHOUSE = pipeline_wh
  AFTER transform_a, transform_b
AS
  INSERT INTO fact_sales SELECT ... FROM dim_customers JOIN dim_products;

-- Resume: children first, then root (bottom-up)
ALTER TASK aggregate_task RESUME;
ALTER TASK transform_a RESUME;
ALTER TASK transform_b RESUME;
ALTER TASK ingest_task RESUME;  -- Root task last

-- Suspend: root first (top-down)
ALTER TASK ingest_task SUSPEND;  -- Root task first
-- Children will not run since root is suspended

-- Serverless task (no warehouse specified)
CREATE TASK cleanup_task
  USER_TASK_MANAGED_INITIAL_WAREHOUSE_SIZE = 'XSMALL'
  SCHEDULE = 'USING CRON 0 0 * * * UTC'
AS
  DELETE FROM temp_data WHERE created_at < DATEADD('day', -7, CURRENT_DATE());

-- WHEN clause with stream check
CREATE TASK process_changes
  WAREHOUSE = pipeline_wh
  SCHEDULE = '1 MINUTE'
  WHEN SYSTEM$STREAM_HAS_DATA('orders_stream')
AS
  MERGE INTO orders_summary ...;

-- WHEN clause with custom condition
CREATE TASK monthly_report
  WAREHOUSE = report_wh
  SCHEDULE = 'USING CRON 0 0 * * * UTC'
  WHEN DAY(CURRENT_DATE()) = 1
AS
  CALL generate_monthly_report();

-- Configure error notifications
CREATE NOTIFICATION INTEGRATION task_errors
  TYPE = QUEUE
  NOTIFICATION_PROVIDER = AWS_SNS
  DIRECTION = OUTBOUND
  AWS_SNS_TOPIC_ARN = 'arn:aws:sns:us-east-1:123456789:task-errors'
  AWS_SNS_ROLE_ARN = 'arn:aws:iam::123456789:role/sns-role'
  ENABLED = TRUE;

-- Set error integration on task
ALTER TASK ingest_task SET ERROR_INTEGRATION = task_errors;

-- Configure retry
ALTER TASK ingest_task SET
  SUSPEND_TASK_AFTER_NUM_FAILURES = 3;

-- Create stream
CREATE STREAM orders_stream ON TABLE orders;

-- View current changes
SELECT * FROM orders_stream;

-- Consume changes (advances offset on commit)
BEGIN;
  INSERT INTO orders_archive
    SELECT order_id, amount, CURRENT_TIMESTAMP()
    FROM orders_stream
    WHERE METADATA$ACTION = 'INSERT';
COMMIT;
-- Stream offset advances after commit

-- Stream is now empty (changes consumed)
SELECT * FROM orders_stream;  -- Returns 0 rows

-- Check stream staleness
SELECT name, stale, stale_after
FROM INFORMATION_SCHEMA.STREAMS
WHERE name = 'ORDERS_STREAM';

-- Stream on a view (Enterprise edition required)
CREATE VIEW active_orders AS
  SELECT * FROM orders WHERE status = 'ACTIVE';

CREATE STREAM active_orders_stream ON VIEW active_orders;

-- Source stream
CREATE STREAM new_events ON TABLE raw_events APPEND_ONLY = TRUE;

-- Scheduled task to process new events
CREATE TASK process_events
  WAREHOUSE = etl_wh
  SCHEDULE = '5 MINUTE'
  WHEN SYSTEM$STREAM_HAS_DATA('new_events')
AS
  INSERT INTO processed_events
  SELECT event_id, event_type, payload,
         PARSE_JSON(payload):user_id::INTEGER AS user_id,
         CURRENT_TIMESTAMP() AS processed_at
  FROM new_events;

-- Track all changes on source
CREATE STREAM customer_changes ON TABLE customers;

-- Task implements SCD Type 2
CREATE TASK update_customer_dim
  WAREHOUSE = etl_wh
  SCHEDULE = '10 MINUTE'
  WHEN SYSTEM$STREAM_HAS_DATA('customer_changes')
AS
  MERGE INTO dim_customer t
  USING (
    SELECT *, METADATA$ACTION AS action, METADATA$ISUPDATE AS is_update
    FROM customer_changes
  ) s
  ON t.customer_id = s.customer_id AND t.is_current = TRUE
  WHEN MATCHED AND s.is_update THEN
    UPDATE SET t.is_current = FALSE, t.end_date = CURRENT_TIMESTAMP()
  WHEN NOT MATCHED AND s.action = 'INSERT' THEN
    INSERT (customer_id, name, email, start_date, end_date, is_current)
    VALUES (s.customer_id, s.name, s.email, CURRENT_TIMESTAMP(), NULL, TRUE);

-- Recent task executions
SELECT name, state, error_code, error_message,
       query_start_time, completed_time,
       DATEDIFF('second', query_start_time, completed_time) AS duration_seconds
FROM TABLE(INFORMATION_SCHEMA.TASK_HISTORY(
  SCHEDULED_TIME_RANGE_START => DATEADD('day', -1, CURRENT_TIMESTAMP()),
  RESULT_LIMIT => 100
))
ORDER BY scheduled_time DESC;

-- Task run states
-- SUCCEEDED: completed successfully
-- FAILED: execution error
-- SKIPPED: WHEN condition was FALSE
-- CANCELLED: task was suspended during execution

-- Check all streams
SELECT name, table_name, type, stale, stale_after, owner
FROM INFORMATION_SCHEMA.STREAMS;

-- Check if stream has data (lightweight check)
SELECT SYSTEM$STREAM_HAS_DATA('my_stream');