TopolearnIncremental pipelines built inside Snowflake avoid reprocessing whole tables and external schedulers.
Three features combine into incremental pipelines.
Streams record changes (CDC) to a table since the stream was last consumed: CREATE STREAM orders_stream ON TABLE orders. Selecting from a stream returns changed rows plus METADATA$ACTION (INSERT or DELETE), METADATA$ISUPDATE and METADATA$ROW_ID; an update appears as a DELETE plus INSERT pair. Reading a stream in a plain SELECT does not advance it. The offset moves only when the stream is used in a DML statement that commits. APPEND_ONLY = TRUE tracks inserts only and is cheaper. A stream that isn't consumed within the table's retention period becomes stale.
Tasks run SQL on a schedule (SCHEDULE = '5 MINUTE' or 'USING CRON 0 * * * * UTC') or after another task (AFTER parent), forming DAGs. Omit WAREHOUSE for a serverless task. A WHEN SYSTEM$STREAM_HAS_DATA('s') condition skips runs when there's nothing to do. Tasks are created suspended; run ALTER TASK ... RESUME.
MERGE applies the changes to a target, handling inserts, updates and deletes in one statement, the same idea as an upsert.
Snowpipe loads files automatically as they land in a stage, triggered by cloud storage event notifications (AUTO_INGEST = TRUE) or a REST call, using serverless compute. It runs a COPY INTO definition and keeps its own load history. Snowpipe Streaming writes rows directly from clients such as Kafka connectors, with no files.
Example
CREATE STREAM orders_stream ON TABLE raw_orders;
CREATE TASK merge_orders
WAREHOUSE = etl_wh
SCHEDULE = '5 MINUTE'
WHEN SYSTEM$STREAM_HAS_DATA('ORDERS_STREAM')
AS
MERGE INTO orders t
USING (SELECT * FROM orders_stream WHERE METADATA$ACTION = 'INSERT') s
ON t.id = s.id
WHEN MATCHED THEN UPDATE SET t.status = s.status, t.amount = s.amount
WHEN NOT MATCHED THEN INSERT (id, status, amount) VALUES (s.id, s.status, s.amount);
ALTER TASK merge_orders RESUME; -- tasks start suspended
-- Continuous file ingestion from an external stage
CREATE PIPE orders_pipe AUTO_INGEST = TRUE AS
COPY INTO raw_orders FROM @s3_orders_stage FILE_FORMAT = (TYPE = JSON)
MATCH_BY_COLUMN_NAME = CASE_INSENSITIVE;Remember
- A stream's offset advances only when it's consumed by committed DML
- Tasks are created suspended; WHEN SYSTEM$STREAM_HAS_DATA avoids empty runs
- Snowpipe auto-ingests files; Snowpipe Streaming ingests rows
Watch out for
- Forgetting ALTER TASK ... RESUME and wondering why nothing runs.