SYSTEM$STREAM_HAS_DATA is a Snowflake task condition that checks whether a named stream has change data available. Used in a task WHEN clause, it allows the task to run only when new or changed data has arrived; otherwise the task is skipped because there is nothing to process. section 13.1.3 section 13.6
Use the condition for incremental pipeline steps that read from streams. A child task can run AFTER a predecessor while independently checking its input stream, avoiding staging or warehouse inserts when no relevant changes are available. Multiple children can use separate stream checks after the same root task. section 13.1.3 section 13.3.2
create task INSERT_ORDERS_STG_TASK
warehouse = 'BAKERY_WH'
after COPY_ORDERS_TASK
when
system$stream_has_data('EXT.JSON_ORDERS_STREAM')
as
insert into STG.JSON_ORDERS_TBL_STG
select
customer_orders:"Customer"::varchar as customer,
customer_orders:"Order date"::date as order_date,
CO.value:"Delivery date"::date as delivery_date,
DO.value:"Baked good type":: varchar as baked_good_type,
DO.value:"Quantity"::number as quantity,
source_file_name,
load_ts
from EXT.JSON_ORDERS_STREAM,
lateral flatten (input => customer_orders:"Orders") CO,
lateral flatten (input => CO.value:"Orders by day") DO;
This dependent task runs after COPY_ORDERS_TASK, then inserts flattened order changes into a staging table only when EXT.JSON_ORDERS_STREAM contains data. section 13.1.3
Verify the condition as part of task operations:
- Resume both the scheduled parent and dependent task after creation so the pipeline can execute. section 13.1.3
- Inspect TASK_HISTORY() when a task is skipped; a stream-reading task with this condition may have had no data available. section 13.5
- Test dependent tasks carefully: tasks defined with AFTER cannot be manually executed unless the dependency is removed and later restored. section 13.1.3
Loading product and partner changes
After PIPELINE_START_TASK, INSERT_PRODUCT_TASK and INSERT_PARTNER_TASK each use WHEN SYSTEM$STREAM_HAS_DATA on their respective staging streams, then insert only rows marked with METADATA$ACTION = 'INSERT' into warehouse tables. section 13.3.2
