Postgres to Snowflake with CDC: Open Source, No Kafka, No Stage

Most Postgres to Snowflake pipelines are a cron job: dump a table every hour, push the file to a stage, COPY INTO a Snowflake table. That is fine for a daily report. It breaks as soon as you need deletes, the latest version of a row within minutes, or the same data in a second engine without paying to copy it again.

There are four common ways to move Postgres data into Snowflake:

• Manual export: COPY to S3, then COPY INTO. Cheap, but full reloads only, and deletes are lost.
• Debezium and Kafka with the Snowflake Kafka connector: seconds of latency, but you run Kafka, Kafka Connect and a schema registry.
• Managed ELT (Fivetran, Estuary, Airbyte Cloud): nothing to run, but you pay per row or GB to land a second copy inside Snowflake.
• Snowflake Openflow: Snowflake's own CDC connector, a good fit when Snowflake is the only consumer.

This walkthrough builds a fifth option. Change data capture (CDC) reads the Postgres write-ahead log with OLake, an open source (Apache 2.0) replication tool, and writes the rows as Iceberg tables straight into Snowflake's own catalog. Snowflake queries them as native managed tables, with no stage, no Snowpipe and no refresh job. The same tables stay readable from DuckDB or Spark.

I ran every step below against a real Snowflake account. The numbers are from that run.

What you'll build

• Source: Postgres 16 in Docker with logical replication on, and two small tables (customers, orders)
• Pipeline: OLake running locally with Docker Compose. It takes a full snapshot first, then replays the write-ahead log
• Destination: Snowflake, with the tables landing as Snowflake-managed tables you query in Snowsight

You need: Docker, a Snowflake account where you can use ACCOUNTADMIN (a trial works), and about 30 minutes.

Step 1: Postgres with logical replication

Save this as docker-compose.yml in an empty folder:

services:
  postgres:
    image: postgres:16
    container_name: pg_cdc
    environment:
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: postgres
      POSTGRES_DB: shop
    command: ["postgres", "-c", "wal_level=logical", "-c", "max_replication_slots=10",
              "-c", "max_wal_senders=10", "-c", "wal_sender_timeout=0"]
    ports:
      - "5432:5432"

docker compose up -d

Now create the tables, a replication user, a publication and a slot. Run this with psql -h localhost -U postgres -d shop:

CREATE TABLE customers (id serial PRIMARY KEY, name text, email text, city text, tier text,
                        created_at timestamptz DEFAULT now(), updated_at timestamptz DEFAULT now());
CREATE TABLE orders (id serial PRIMARY KEY, customer_id int REFERENCES customers(id),
                     product text, amount numeric(10,2), status text,
                     created_at timestamptz DEFAULT now(), updated_at timestamptz DEFAULT now());

INSERT INTO customers (name, email, city, tier) VALUES
  ('Aarav Mehta','[e-mail]','Mumbai','gold'),   ('Priya Sharma','[e-mail]','Bengaluru','silver'),
  ('Liam Carter','[e-mail]','Austin','bronze'),  ('Sofia Rossi','[e-mail]','Milan','gold'),
  ('Kenji Tanaka','[e-mail]','Tokyo','silver'), ('Emma Schulz','[e-mail]','Berlin','bronze'),
  ('Noah Kim','[e-mail]','Seoul','gold'),        ('Zara Ali','[e-mail]','Dubai','silver');

INSERT INTO orders (customer_id, product, amount, status)
SELECT 1 + (n % 8), (ARRAY['Laptop','Headphones','Monitor','Keyboard','Mouse','Webcam'])[1 + n % 6],
       ROUND(20 + ((n * 7919) % 120000) / 100.0, 2), (ARRAY['placed','shipped','delivered'])[1 + n % 3]
FROM generate_series(1, 40) AS n;

-- The user OLake connects as
CREATE ROLE olake WITH LOGIN REPLICATION PASSWORD 'olake_password';
GRANT USAGE ON SCHEMA public TO olake;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO olake;
ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO olake;

-- What OLake reads changes from
CREATE PUBLICATION olake_pub FOR TABLE customers, orders;
SELECT pg_create_logical_replication_slot('olake_slot', 'pgoutput');

💡 One slot and one publication per pipeline. Two jobs sharing a slot each read part of the other's changes, and both lose data.

A few things that matter on a real database:

• Postgres 10 or later. Logical replication with the built-in pgoutput plugin arrived in 10.
• Tables without a primary key need ALTER TABLE <table> REPLICA IDENTITY FULL;, or their updates and deletes don't carry the old row.
• The slot holds WAL until it's read. Pause the pipeline for a week and Postgres keeps a week of WAL on disk. Set max_slot_wal_keep_size so a stalled slot is dropped before the disk fills, and alert on pg_replication_slots.wal_status.
• wal_sender_timeout=0 stops a long first snapshot from being cut off. It's in the compose file above.

On RDS, Aurora, Cloud SQL or Azure you set wal_level through the parameter group instead of the command line. Everything else is the same.

Step 2: Prepare Snowflake

Run this once in a Snowsight worksheet as ACCOUNTADMIN. It creates the database the tables land in, a role and a service user for OLake, and the token OLake logs in with.

USE ROLE ACCOUNTADMIN;
CREATE WAREHOUSE IF NOT EXISTS OLAKE_WH WAREHOUSE_SIZE = XSMALL AUTO_SUSPEND = 60 AUTO_RESUME = TRUE;
CREATE DATABASE IF NOT EXISTS OLAKE_DEMO;
ALTER DATABASE OLAKE_DEMO SET CATALOG = 'SNOWFLAKE';
ALTER DATABASE OLAKE_DEMO SET EXTERNAL_VOLUME = 'SNOWFLAKE_MANAGED';

CREATE ROLE IF NOT EXISTS OLAKE_ROLE;
GRANT USAGE, CREATE SCHEMA ON DATABASE OLAKE_DEMO TO ROLE OLAKE_ROLE;
GRANT USAGE ON WAREHOUSE OLAKE_WH TO ROLE OLAKE_ROLE;
GRANT ROLE OLAKE_ROLE TO ROLE SYSADMIN;
CREATE USER IF NOT EXISTS OLAKE_SVC TYPE = SERVICE DEFAULT_ROLE = OLAKE_ROLE;
GRANT ROLE OLAKE_ROLE TO USER OLAKE_SVC;

-- A token for a service user needs a network policy. Allow the IP OLake connects from.
USE SCHEMA OLAKE_DEMO.PUBLIC;
CREATE NETWORK RULE IF NOT EXISTS olake_ip MODE = INGRESS TYPE = IPV4 VALUE_LIST = ('<your public IP>/32');
CREATE NETWORK POLICY IF NOT EXISTS olake_network_policy ALLOWED_NETWORK_RULE_LIST = ('OLAKE_DEMO.PUBLIC.olake_ip');
ALTER USER OLAKE_SVC SET NETWORK_POLICY = olake_network_policy;

SELECT CURRENT_ORGANIZATION_NAME() || '-' || CURRENT_ACCOUNT_NAME() AS account_identifier, CURRENT_REGION() AS region;
ALTER USER OLAKE_SVC ADD PROGRAMMATIC ACCESS TOKEN OLAKE_PAT ROLE_RESTRICTION = 'OLAKE_ROLE' DAYS_TO_EXPIRY = 30;

curl -s https://checkip.amazonaws.com prints your public IP. Keep the account_identifier, the region and the token from the last statement. Snowflake shows the token only once.

⚠️ Skip the network policy and the last line fails with Network Policy is required when creating a programmatic access token for user OLAKE_SVC.

The two ALTER DATABASE lines do the important part. They make every table OLake creates a Snowflake-managed table on Snowflake's own storage. That is why there is no stage, no external volume and no CREATE TABLE later.

Step 3: Run OLake

curl -sSL https://raw.githubusercontent.com/datazip-inc/olake-ui/master/docker-compose-v1.yml | docker compose -f - up -d

Open http://localhost:8000 and sign in with admin / password.

Source. Go to Sources, then Create Source, then Postgres. Leave Enable CDC on and enter olake_slot as the Replication Slot and olake_pub as the Publication. For the connection, use host host.docker.internal (the containers reach your machine through it), port 5432, database shop, user olake and password olake_password. Click Create. OLake checks wal_level, the replication privilege, the slot and the publication, and tells you which one is missing.

Destination. Go to Destinations, then Create Destination, then Apache Iceberg, then catalog type Horizon, then Programmatic Access Token (PAT):

Field	Value
Horizon Iceberg REST Catalog URI	https://<account_identifier>.snowflakecomputing.com/polaris/api/catalog
Snowflake Database	OLAKE_DEMO
Programmatic Access Token	the token from step 2
Session Role Scope	session:role:OLAKE_ROLE
Cloud Storage Region	the region from step 2, for example AWS_AP_SOUTH_1

Job. Go to Jobs, then Create Job. Name it pg_to_snowflake, pick the source and destination, and under Advanced Settings set Target Query Engines to Snowflake, plus DuckDB and Apache Spark if they'll read the data too. That setting matters: OLake picks the format for updates and deletes that every listed engine can read. Leave it empty and Snowflake won't see your updates. On the Streams page tick both tables and keep Full Refresh + CDC. Then click Create Job and Sync Now.

Step 4: Query it in Snowflake

In my run the first sync copied 8 customers and 40 orders in 41 seconds, container start included. The tables show up right away:

SHOW ICEBERG TABLES IN SCHEMA OLAKE_DEMO."pg_to_snowflake_shop_public";

SELECT "id", "name", "city", "tier", "_op_type"
FROM OLAKE_DEMO."pg_to_snowflake_shop_public"."customers"
ORDER BY "id";

SHOW ICEBERG TABLES reports iceberg_table_type = MANAGED. Snowflake owns these tables, and there is nothing to refresh.

⚠️ Quote the names. OLake keeps the lowercase Postgres names, and Snowflake upper-cases anything unquoted. That turns into Object 'OLAKE_DEMO.PG_TO_SNOWFLAKE_SHOP_PUBLIC.CUSTOMERS' does not exist or not authorized.

Step 5: Change rows and sync again

This is the part a cron job gets wrong:

INSERT INTO customers (name, email, city, tier) VALUES
  ('Maya Patel','[e-mail]','Pune','silver'), ('Lucas Silva','[e-mail]','São Paulo','bronze');
INSERT INTO orders (customer_id, product, amount, status) VALUES
  (9,'Laptop',1299.00,'placed'), (9,'Mouse',25.50,'placed'), (10,'Monitor',349.99,'placed'),
  (3,'Keyboard',89.00,'placed'), (7,'Webcam',79.90,'placed');
UPDATE customers SET tier = 'gold',      updated_at = now() WHERE name = 'Liam Carter';
UPDATE customers SET city = 'Abu Dhabi', updated_at = now() WHERE name = 'Zara Ali';
UPDATE orders SET status = 'delivered',  updated_at = now() WHERE id IN (2, 5, 8);

Run Sync Now again. The second sync read 12 change events in 23 seconds from the replication slot. For customers it wrote one data file with 4 rows (the 2 inserts and the new versions of the 2 updated rows), plus one small delete file that retires the 2 old versions.

Re-run the SELECT: Liam is gold, Zara is in Abu Dhabi, and _op_type shows c for the inserts and u for the updates.

Deletes work the same way. DELETE FROM customers WHERE id = 6; and a sync leave a row with _op_type = 'd', so filter with WHERE "_op_type" <> 'd' for live rows.

Schema changes. When you add a column in Postgres, OLake adds it to the table on the next sync and Snowflake shows it. Widening changes (int to bigint) are applied. Narrowing or incompatible changes fail the sync instead of corrupting the table.

Read the same tables from DuckDB

The tables are open Apache Iceberg tables in Snowflake's catalog, so another engine can read the same files without an export or a second copy. DuckDB wants a short-lived bearer token instead of the raw access token, so exchange it first:

curl -s -X POST "https://<account_identifier>.snowflakecomputing.com/polaris/api/catalog/v1/oauth/tokens" \
  -H 'Content-Type: application/x-www-form-urlencoded' \
  --data-urlencode grant_type=client_credentials \
  --data-urlencode scope=session:role:OLAKE_ROLE \
  --data-urlencode client_secret=<your token>

INSTALL iceberg; LOAD iceberg;
INSTALL httpfs;  LOAD httpfs;
CREATE SECRET horizon_secret (TYPE iceberg, TOKEN '<access_token from the curl call>');
ATTACH 'OLAKE_DEMO' AS horizon (
  TYPE iceberg,
  ENDPOINT 'https://<account_identifier>.snowflakecomputing.com/polaris/api/catalog',
  SECRET horizon_secret,
  ACCESS_DELEGATION_MODE 'vended_credentials'
);
SELECT id, name, city, tier FROM horizon.pg_to_snowflake_shop_public.customers ORDER BY id;

DuckDB 1.5.5 returned the same 10 rows as Snowflake, and Spark 3.5.6 did too. An orders roll-up by status matched Postgres to the cent in all three engines. The DuckDB path needs a Snowflake account on AWS or GCP. The token is valid for about an hour.

Data types to know about

Postgres	Snowflake shows	Note
int, serial	NUMBER(10,0)
bigint	NUMBER(19,0)
numeric, decimal	FLOAT	stored as doubles; round when you need exact cents
timestamp, timestamptz, date	TIMESTAMP_LTZ(6)	UTC
jsonb, uuid, arrays	VARCHAR	use PARSE_JSON()

In my run the orders.amount totals still matched Postgres to the cent.

Errors you might hit

FATAL: must be superuser or replication role to start walsender. The OLake user can't replicate. Run ALTER ROLE olake WITH REPLICATION;. On RDS and Aurora, grant rds_replication instead.

Object '...CUSTOMERS' does not exist or not authorized. The name wasn't quoted; see step 4.

The destination test fails. The token exchange with Snowflake is failing. Check, in order:

1. The service user has the network policy, and OLake's IP is in the rule.
2. The scope is session:role:OLAKE_ROLE, matching the token's ROLE_RESTRICTION.
3. The token hasn't expired. An expired token fails with Programmatic access token is expired.; add a new one.

Running the curl exchange above shows the same error without the UI in the way.

Snowflake shows stale rows after updates. The job was created without a target query engine, so it wrote a delete format Snowflake can't read. Set Snowflake as a target engine on the job, and the next sync rewrites the delete files.

How this compares

Method	Freshness	Deletes	Where the data lands	What you run
COPY to S3 plus COPY INTO	hours to daily	no, full reloads	Snowflake tables	a script and a scheduler
Debezium, Kafka and the Snowflake connector	seconds	yes	Kafka, then Snowflake tables	Kafka, Connect, Debezium, schema registry
Fivetran, Estuary, Airbyte Cloud	minutes, by plan	yes	Snowflake tables (a second copy)	nothing; per-row or per-GB pricing
Snowflake Openflow	CDC	yes, as soft deletes	native Snowflake tables	an Openflow runtime
OLake to Snowflake's catalog (this post)	CDC, schedule down to 1 minute	yes	Snowflake-managed tables that DuckDB and Spark can read too	one Docker Compose stack, open source

If Snowflake is the only thing that will ever read this data, Openflow or a managed tool is the shorter path. The approach above pays off when a second engine needs the same rows, or you want your data in an open format instead of paying to copy it around.

Where to go from here

• Point the source at your real database. RDS, Aurora, Cloud SQL and Azure only differ in how you set wal_level.
• Add tables with ALTER PUBLICATION olake_pub ADD TABLE ..., then tick them in the job.
• Want the files in your own S3 bucket instead of Snowflake storage? Use AWS Glue as the catalog. Snowflake then reads the tables through a catalog integration, and Athena and EMR can read them too.
• The full guide, with the Glue setup and the Spark connection, is on olake.io.

If a step doesn't match what you see, say so in the comments and I'll update the post.