Overview

What You'll Build

What You'll Learn

What You'll Need

How It Works

Architecture

Stream Change Record Semantics

File Ordering Guarantee

Data Model

Prerequisites

Postgres Instance Setup

Step 1: Create Network Policy

-- Postgres Instance Setup: Step 1 - Create Network Policy
-- Execute in: Snowsight (Snowflake)
USE ROLE ACCOUNTADMIN;

CREATE DATABASE IF NOT EXISTS pg_network_db;
CREATE SCHEMA  IF NOT EXISTS pg_network_db.pg_network;

USE SCHEMA pg_network_db.pg_network;

-- Run this to find your current IP address if needed
SELECT current_ip_address();

CREATE OR REPLACE NETWORK RULE pg_lab_ingress_rule
    TYPE       = IPV4
    VALUE_LIST = ('nnn.nnn.nnn.nnn/32')
    MODE       = POSTGRES_INGRESS
    COMMENT    = 'Allow Postgres client connections (restrict in production)';

CREATE OR REPLACE NETWORK POLICY pg_lab_network_policy
    ALLOWED_NETWORK_RULE_LIST = ('pg_lab_ingress_rule')
    COMMENT = 'Network policy for Snowflake Postgres instances';

Step 2: Create Postgres Instance

-- Postgres Instance Setup: Step 2 - Create Postgres Instance
-- Execute in: Snowsight (Snowflake)

CREATE POSTGRES INSTANCE PG_LAB
    AUTHENTICATION_AUTHORITY = POSTGRES_OR_SNOWFLAKE
    COMPUTE_FAMILY           = 'STANDARD_L'
    STORAGE_SIZE_GB          = 50
    POSTGRES_VERSION         = 18
    HIGH_AVAILABILITY        = FALSE
    NETWORK_POLICY           = 'pg_lab_network_policy'
    COMMENT                  = 'Postgres instance for stream-to-postgres quickstart';

Important: Save the username and password displayed in the access_roles field and the host value — you will need them to connect.

Step 3: Monitor Instance Status

-- Postgres Instance Setup: Step 3 - Monitor Instance Status
-- Execute in: Snowsight (Snowflake)

SHOW POSTGRES INSTANCES;

Step 4: Configure psql Connection

export PGHOST=<hostname from SHOW POSTGRES INSTANCES>
export PGPORT=5432
export PGDATABASE=postgres
export PGUSER=snowflake_admin
export PGPASSWORD=<password from CREATE POSTGRES INSTANCE output>
export PGSSLMODE=require

psql -c "SELECT version();"

Snowflake Setup

Role Required: This section requires ACCOUNTADMIN to create the storage integration.

Step 1: Create Database, Schema, Products Table, and Sample Data

-- Snowflake Setup: Step 1 - Create Database, Schema, Products Table, and Sample Data
-- Execute in: Snowsight (Snowflake)

CREATE DATABASE IF NOT EXISTS stream_lab;
CREATE SCHEMA  IF NOT EXISTS stream_lab.catalog;

USE SCHEMA stream_lab.catalog;

CREATE OR REPLACE TABLE products (
    product_id    INT            NOT NULL,
    product_name  VARCHAR(100)   NOT NULL,
    category      VARCHAR(50)    NOT NULL,
    price         NUMBER(10,2)   NOT NULL,
    status        VARCHAR(20)    NOT NULL,
    updated_at    TIMESTAMP_NTZ  DEFAULT current_timestamp(),  -- must be set on every INSERT and UPDATE; used for sync ordering
    PRIMARY KEY (product_id)
);

INSERT INTO products (product_id, product_name, category, price, status)
VALUES
    (1, 'Wireless Headphones',   'electronics',  89.99, 'active'),
    (2, 'USB-C Hub 7-Port',      'electronics',  49.99, 'active'),
    (3, 'Mechanical Keyboard',   'electronics', 129.99, 'active'),
    (4, 'Running Jacket',        'apparel',       74.99, 'active'),
    (5, 'Merino Wool Socks',     'apparel',       18.99, 'active'),
    (6, 'Trail Running Shoes',   'apparel',      119.99, 'active'),
    (7, 'Bamboo Desk Organizer', 'home',          34.99, 'active'),
    (8, 'LED Desk Lamp',         'home',          55.99, 'active');

Step 2: Create Storage Integration, Stage, and Verify

-- Snowflake Setup: Step 2 - Create Storage Integration, Stage, and Verify
-- Execute in: Snowsight (Snowflake)
CREATE OR REPLACE STORAGE INTEGRATION pg_lab_stream_int
    TYPE              = POSTGRES_INTERNAL_STORAGE
    POSTGRES_INSTANCE = 'PG_LAB'
    ENABLED           = TRUE;

CREATE OR REPLACE STAGE stream_sync_stage
    STORAGE_INTEGRATION = pg_lab_stream_int
    RELATIVE_URL        = '/';

-- Confirm stage is accessible (returns empty result if no files yet)
LIST @stream_sync_stage/;

Step 3: Create Stream and View Initial Rows

-- Snowflake Setup: Step 3 - Create Stream and View Initial Rows
-- Execute in: Snowsight (Snowflake)
CREATE OR REPLACE STREAM products_stream
    ON TABLE products
    APPEND_ONLY       = FALSE
    SHOW_INITIAL_ROWS = TRUE
    COMMENT = 'Captures all changes to the products table for sync to Postgres';

-- All 8 existing rows should appear with _action = 'INSERT', _is_update = FALSE
SELECT
    METADATA$ACTION   AS _action,
    METADATA$ISUPDATE AS _is_update,
    product_id, product_name, category, price, status
FROM products_stream
ORDER BY product_id;

Step 4: Create Export Procedure, Task, and Verify

-- Snowflake Setup: Step 4 - Create Export Procedure, Task, and Verify
-- Execute in: Snowsight (Snowflake)
CREATE OR REPLACE PROCEDURE export_products_stream()
RETURNS STRING
LANGUAGE SQL
EXECUTE AS CALLER
AS
$$
DECLARE
    stage_path TEXT;
    row_count  INTEGER;
BEGIN
    row_count := (SELECT COUNT(*) FROM stream_lab.catalog.products_stream);

    IF (row_count = 0) THEN
        RETURN NULL;
    END IF;

    stage_path := '@stream_sync_stage/products/product_stream_'
                  || TO_CHAR(CURRENT_TIMESTAMP(), 'YYYYMMDD_HH24MISS')
                  || '.parquet';

    EXECUTE IMMEDIATE
        'COPY INTO ' || :stage_path || '
        FROM (
            SELECT
                product_id,
                product_name,
                category,
                ROUND(price * 100)::INTEGER AS price_cents,
                status,
                updated_at,
                METADATA$ACTION            AS _action,
                METADATA$ISUPDATE::BOOLEAN AS _is_update
            FROM stream_lab.catalog.products_stream
        )
        FILE_FORMAT = (TYPE = ''PARQUET'')
        SINGLE      = TRUE';

    RETURN :stage_path || ' (' || :row_count::TEXT || ' rows)';
END;
$$;

CREATE OR REPLACE TASK export_products_stream_task
    WAREHOUSE = COMPUTE_WH
    SCHEDULE  = '1 MINUTE'
WHEN
    SYSTEM$STREAM_HAS_DATA('stream_lab.catalog.products_stream')
AS
    CALL export_products_stream();

ALTER TASK export_products_stream_task RESUME;

-- Wait up to 1 minute for the task to run on schedule, then verify
-- Files land at products/product_stream_YYYYMMDD_HH24MISS.parquet
LIST @stream_sync_stage/products/;

-- Confirm the stream is now empty (task consumed the initial rows)
SELECT count(*) AS pending_changes FROM products_stream;

Postgres Setup

Step 1: Enable Extensions

psql

-- Postgres Setup: Step 1 - Enable Extensions
-- Execute in: psql (Postgres)
CREATE EXTENSION IF NOT EXISTS pg_lake CASCADE;
CREATE EXTENSION IF NOT EXISTS pg_cron;
CREATE EXTENSION IF NOT EXISTS pg_incremental;

Step 2: Create Products Table

-- Postgres Setup: Step 2 - Create Products Table
-- Execute in: psql (Postgres)
CREATE TABLE IF NOT EXISTS products (
    product_id    INTEGER        NOT NULL PRIMARY KEY,
    product_name  TEXT           NOT NULL,
    category      TEXT           NOT NULL,
    price         NUMERIC(10,2)  NOT NULL,
    status        TEXT           NOT NULL,
    updated_at    TIMESTAMP      NOT NULL,
    synced_at     TIMESTAMP      NOT NULL DEFAULT now()
);

Step 3: Create Foreign Table

Note on monetary values: Snowflake exports NUMBER(10,2) columns as DOUBLE PRECISION in Parquet format (IEEE 754 binary float). Most decimal fractions — including common prices like 89.99 — cannot be represented exactly in binary floating point. While PostgreSQL's shortest-representation conversion handles most cases correctly, this is not guaranteed for all possible values. To eliminate this risk entirely, the Snowflake task exports price as integer cents (price * 100) and the sync function divides by 100 when inserting into Postgres. This keeps monetary values as exact integers throughout the pipeline.

Format comparison for fixed-point values:

Format	How NUMBER(10,2) travels	Precision risk	Foreign table type
Parquet	Exported as DOUBLE (float64)	Yes — cents approach required	INTEGER (cents)
CSV	Exported as decimal string "89.99"	None — exact text round-trip	NUMERIC(10,2) directly
JSON	Exported as decimal string 89.99, but many JSON parsers parse numbers as float64 internally	Depends on pg_lake implementation	Test before trusting NUMERIC

If you switch to CSV format, you can remove the cents conversion entirely — declare price NUMERIC(10,2) in the foreign table and remove the price_cents column and division from the sync function.

-- Postgres Setup: Step 3 - Create Foreign Table
-- Execute in: psql (Postgres)
CREATE FOREIGN TABLE products_stream_files (
    product_id    INTEGER,
    product_name  TEXT,
    category      TEXT,
    price_cents   INTEGER,
    status        TEXT,
    updated_at    TIMESTAMP,
    _action       TEXT,
    _is_update    BOOLEAN,
    _filename     TEXT
)
SERVER pg_lake
OPTIONS (
    path     '@STAGE/products/product_stream_*.parquet',
    format   'parquet',
    filename 'true'
);

Step 4: Create Sync Function

How Snowflake streams work — net outcome, not history: A standard (delta) stream does not replay every DML statement applied to the source. It exposes the net change of each row between two offsets (the last time the stream was consumed and now). All intermediate churn for a given key is collapsed before you ever see it. Concretely, within a single consume — and therefore within a single exported file — each product_id appears as at most one of the following:

Records for a key in one file	METADATA$ISUPDATE (_is_update)	Net meaning
INSERT alone	false	net-new row
DELETE alone	false	genuine removal
DELETE + INSERT pair	true (both)	update

You can never get two independent inserts, or an unrelated delete and insert, for the same key in one file. For example, if a row is deleted and then re-inserted before the next consume, the stream nets it to a single INSERT (if the row was absent at the start offset) or to a DELETE+INSERT update pair (if it was present) — never to two separate events. This is why a per-key final-state resolution is always well defined.

Note on updated_at: Because routing is decided by record structure (_is_update) rather than by ordering, updated_at is not a control key for this function — it is stored as a plain data column. Cross-file ordering is handled separately by the sorted list wrapper in Step 5 (file modification time), which sequences one file after another. Within a file there is only one net state per key, so no per-record ordering is required. Snowflake streams expose only three metadata columns — METADATA$ACTION, METADATA$ISUPDATE, and METADATA$ROW_ID — and _is_update (METADATA$ISUPDATE) is the one that makes this deterministic.

-- Postgres Setup: Step 4 - Create Sync Function
-- Execute in: psql (Postgres)
CREATE OR REPLACE FUNCTION sync_products_from_file(filepath TEXT)
RETURNS VOID
LANGUAGE SQL AS $$
    MERGE INTO products t
    USING (
        SELECT product_id, product_name, category, (price_cents::NUMERIC / 100)::NUMERIC(10,2) AS price,
               status, updated_at::TIMESTAMP AS updated_at, _action
        FROM   products_stream_files
        WHERE  _filename = filepath
               AND  NOT (_action = 'DELETE' AND _is_update = true)  -- drop the delete-half of updates
    ) s
    ON t.product_id = s.product_id
    WHEN MATCHED AND s._action = 'DELETE' THEN
        DELETE
    WHEN MATCHED AND s._action = 'INSERT' THEN
        UPDATE SET product_name = s.product_name,
                   category     = s.category,
                   price        = s.price,
                   status       = s.status,
                   updated_at   = s.updated_at,
                   synced_at    = now()
    WHEN NOT MATCHED AND s._action = 'INSERT' THEN
        INSERT (product_id, product_name, category, price, status, updated_at, synced_at)
        VALUES (s.product_id, s.product_name, s.category, s.price, s.status, s.updated_at, now());
$$;

Step 5: Create Sorted List Wrapper and pg_incremental Pipeline

-- Postgres Setup: Step 5 - Create Sorted List Wrapper and pg_incremental Pipeline
-- Execute in: psql (Postgres)

-- Sorted wrapper: returns files ordered by modification time ascending.
-- pg_incremental calls this with one text argument (the file pattern) and
-- reads only the path column, so additional columns are harmless.
CREATE OR REPLACE FUNCTION public.list_files_sorted(url_wildcard text,
    OUT path text,
    OUT file_size bigint,
    OUT last_modified_time timestamptz,
    OUT etag text)
RETURNS SETOF record
LANGUAGE SQL STABLE AS $$
    SELECT path, file_size, last_modified_time, etag
    FROM   lake_file.list(url_wildcard)
    ORDER BY last_modified_time ASC NULLS LAST;
$$;

-- Pipeline: polls every 30 seconds, calls sync_products_from_file per new file,
-- and processes files in modification-time order via the sorted wrapper.
SELECT incremental.create_file_list_pipeline(
    pipeline_name => 'products_sync_pipeline',
    file_pattern  => '@STAGE/products/product_stream_*.parquet',
    command       => $$SELECT sync_products_from_file($1)$$,
    list_function => 'public.list_files_sorted',
    schedule      => '30 seconds'
);

Step 6: Monitor the Pipeline

-- Postgres Setup: Step 6 - Monitor the Pipeline
-- Execute in: psql (Postgres)
SELECT   jobname, start_time, end_time, status, return_message
FROM     cron.job_run_details
         JOIN cron.job USING (jobid)
WHERE    jobname LIKE '%products_sync_pipeline%'
ORDER BY start_time DESC
LIMIT    10;

Step 7: Verify Initial Data Loaded

-- Postgres Setup: Step 7 - Verify Initial Data Loaded
-- Execute in: psql (Postgres)
SELECT   product_id, product_name, category, price, status, updated_at, synced_at
FROM     products
ORDER BY product_id;

Step 8: Check Processed Files

-- Postgres Setup: Step 8 - Check Processed Files
-- Execute in: psql (Postgres)
SELECT *
FROM   incremental.processed_files
WHERE  pipeline_name = 'products_sync_pipeline';

Step 9: Schedule Stage File Purge

-- Postgres Setup: Step 9 - Schedule Stage File Purge
-- Execute in: psql (Postgres)

CREATE OR REPLACE FUNCTION purge_old_sync_files(older_than_days INT DEFAULT 7)
RETURNS INT LANGUAGE plpgsql AS $$
DECLARE
    rec           RECORD;
    deleted_count INT := 0;
BEGIN
    FOR rec IN
        SELECT file_path
        FROM   incremental.processed_files
        WHERE  pipeline_name = 'products_sync_pipeline'
          AND  processed_at  < now() - (older_than_days || ' days')::interval
    LOOP
        BEGIN
            -- Stage delete first: if this fails the processed_files row is
            -- preserved and the file will not be reprocessed
            PERFORM lake_file.delete(rec.file_path);
            PERFORM lake_file_cache.remove(rec.file_path);

            -- Only remove the tracking row after the stage delete succeeds
            DELETE FROM incremental.processed_files
            WHERE  file_path = rec.file_path;

            deleted_count := deleted_count + 1;
        EXCEPTION WHEN OTHERS THEN
            RAISE WARNING 'purge_old_sync_files: skipping % — %',
                          rec.file_path, SQLERRM;
        END;
    END LOOP;

    RETURN deleted_count;
END;
$$;

-- Run daily at 03:00 UTC, retaining the last 30 days of files
SELECT cron.schedule(
    'purge-old-sync-files',
    '0 3 * * *',
    $$SELECT purge_old_sync_files(7)$$
);

-- Confirm the job is registered
SELECT jobid, jobname, schedule, active FROM cron.job ORDER BY jobid;

Observe Live Sync

Step 1: Make Changes in Snowflake

-- Observe Live Sync: Step 1 - Make Changes in Snowflake
-- Execute in: Snowsight (Snowflake)
USE SCHEMA stream_lab.catalog;

-- Update: raise product 1 price from 89.99 to 99.99
UPDATE products
SET    price = 99.99, updated_at = current_timestamp()
WHERE  product_id = 1;

-- Insert: add a new product
INSERT INTO products (product_id, product_name, category, price, status)
VALUES (9, 'Smart Water Bottle', 'home', 42.99, 'active');

-- Delete: discontinue product 5
DELETE FROM products WHERE product_id = 5;

Step 2: View Pending Stream Changes

-- Observe Live Sync: Step 2 - View Pending Stream Changes
-- Execute in: Snowsight (Snowflake)
SELECT
    METADATA$ACTION   AS _action,
    METADATA$ISUPDATE AS _is_update,
    product_id, product_name, category, price, status
FROM products_stream
ORDER BY product_id;

Step 3: Wait for Task to Export Changes

-- Observe Live Sync: Step 3 - Wait for Task to Export Changes
-- Execute in: Snowsight (Snowflake)
LIST @stream_sync_stage/products/;

-- Confirm stream is empty (task consumed the changes)
SELECT count(*) AS pending_changes FROM products_stream;

Step 4: Verify Changes in Postgres

-- Observe Live Sync: Step 4 - Verify Changes in Postgres
-- Execute in: psql (Postgres)
SELECT   product_id, product_name, price, status, updated_at, synced_at
FROM     products
ORDER BY product_id;

Cleanup

Postgres Cleanup

-- Cleanup: Postgres Cleanup
-- Execute in: psql (Postgres)
SELECT cron.unschedule('purge-old-sync-files');

DROP FUNCTION      IF EXISTS purge_old_sync_files(INT);

SELECT incremental.drop_pipeline('products_sync_pipeline');

DROP FUNCTION      IF EXISTS sync_products_from_file(TEXT);
DROP FOREIGN TABLE IF EXISTS products_stream_files;
DROP TABLE         IF EXISTS products;

SELECT lake_file_cache.remove(path) FROM lake_file_cache.list();

Snowflake Cleanup

-- Cleanup: Snowflake Cleanup
-- Execute in: Snowsight (Snowflake)
USE ROLE ACCOUNTADMIN;
USE SCHEMA stream_lab.catalog;

ALTER TASK IF EXISTS stream_lab.catalog.export_products_stream_task SUSPEND;
DROP  TASK      IF EXISTS stream_lab.catalog.export_products_stream_task;
DROP  PROCEDURE IF EXISTS stream_lab.catalog.export_products_stream();

DROP STREAM IF EXISTS stream_lab.catalog.products_stream;
DROP TABLE  IF EXISTS stream_lab.catalog.products;

REMOVE @stream_sync_stage/products/;

DROP STAGE               IF EXISTS stream_lab.catalog.stream_sync_stage;
DROP STORAGE INTEGRATION IF EXISTS PG_LAB_stream_int;

DROP SCHEMA   IF EXISTS stream_lab.catalog;
DROP DATABASE IF EXISTS stream_lab;

Infrastructure Cleanup (Optional)

-- Cleanup: Infrastructure Cleanup (Optional)
-- Execute in: Snowsight (Snowflake)
USE ROLE ACCOUNTADMIN;

DROP POSTGRES INSTANCE IF EXISTS PG_LAB;
DROP NETWORK POLICY    IF EXISTS PG_LAB_network_policy;
DROP NETWORK RULE      IF EXISTS pg_network_db.pg_network.PG_LAB_ingress_rule;

Conclusion and Resources

What You Learned

Key Capabilities Demonstrated

Related Resources