Replicate PostgreSQL to Snowflake with CDC

ALTER SYSTEM SET wal_level = logical;

CREATE ROLE filament WITH LOGIN REPLICATION PASSWORD 'change-me';
GRANT USAGE ON SCHEMA public TO filament;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO filament;

CREATE PUBLICATION filament FOR TABLE public.customers, public.orders;

ALTER TABLE public.orders REPLICA IDENTITY FULL;

openssl genrsa 2048 | openssl pkcs8 -topk8 -inform PEM -out filament_key.p8 -nocrypt
openssl rsa -in filament_key.p8 -pubout -out filament_key.pub
grep -v '^-----' filament_key.pub | tr -d '\n'

USE ROLE SECURITYADMIN;
CREATE ROLE IF NOT EXISTS FILAMENT_ROLE;
GRANT ROLE FILAMENT_ROLE TO ROLE SYSADMIN;

USE ROLE SYSADMIN;
CREATE WAREHOUSE IF NOT EXISTS FILAMENT_WAREHOUSE
  WAREHOUSE_SIZE = XSMALL AUTO_SUSPEND = 60 AUTO_RESUME = TRUE INITIALLY_SUSPENDED = TRUE;
CREATE DATABASE IF NOT EXISTS FILAMENT_DATABASE;
GRANT USAGE ON WAREHOUSE FILAMENT_WAREHOUSE TO ROLE FILAMENT_ROLE;
GRANT USAGE, CREATE SCHEMA ON DATABASE FILAMENT_DATABASE TO ROLE FILAMENT_ROLE;

USE ROLE SECURITYADMIN;
CREATE USER IF NOT EXISTS FILAMENT_USER
  TYPE = SERVICE DEFAULT_ROLE = FILAMENT_ROLE DEFAULT_WAREHOUSE = FILAMENT_WAREHOUSE;
GRANT ROLE FILAMENT_ROLE TO USER FILAMENT_USER;
ALTER USER FILAMENT_USER SET BINARY_INPUT_FORMAT = 'BASE64';
ALTER USER FILAMENT_USER SET TIMESTAMP_INPUT_FORMAT = 'AUTO';
ALTER USER FILAMENT_USER SET RSA_PUBLIC_KEY = '<PUBLIC_KEY_BODY>';

filament source create app-db \
  --source-connector postgres \
  --source-connection-method url \
  --source-dsn "postgres://filament:change-me@db.internal:5432/shop?sslmode=require" \
  --source-replication cdc \
  --source-manage-publication=false

filament sink create snowflake \
  --sink-connector snowflake \
  --sink-account myorg-myaccount \
  --sink-username FILAMENT_USER \
  --sink-warehouse FILAMENT_WAREHOUSE \
  --sink-database-name FILAMENT_DATABASE \
  --sink-role FILAMENT_ROLE \
  --sink-auth "$(jq -n --rawfile k filament_key.p8 '{type: "key_pair", private_key: $k}')"

filament source discover app-db

filament pipeline create app-db-to-snowflake \
  --source app-db \
  --sink snowflake \
  --resources customers,orders \
  --sync-mode cdc \
  --write-mode merge \
  --sink-schema SHOP

filament run app-db-to-snowflake

UPDATE orders SET status = 'shipped', updated_at = now() WHERE id BETWEEN 2001 AND 3000;
DELETE FROM orders WHERE id BETWEEN 5001 AND 5200;
INSERT INTO orders SELECT 1000500 + g, 1 + g, 1 + g, 2, 19.99, 'placed', now(), now()
FROM generate_series(1, 300) g;
UPDATE customers SET country = 'CA', updated_at = now() WHERE id = 42;

filament run app-db-to-snowflake

-- PostgreSQL
SELECT count(*) FROM orders;

-- Snowflake
SELECT count(*) FROM FILAMENT_DATABASE.SHOP.ORDERS;
SELECT count(*) FROM FILAMENT_DATABASE.SHOP.ORDERS WHERE id BETWEEN 5001 AND 5200;  -- 0

SELECT slot_name, active,
       pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn)) AS lag
FROM pg_replication_slots;