Replicate PostgreSQL to ClickHouse with Filament

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

filament source create shop-pg \
  --source-connector postgres \
  --source-connection-method url \
  --source-dsn "postgres://filament:change-me@db.internal:5432/shop?sslmode=require"

filament sink create clickhouse \
  --sink-connector clickhouse \
  --sink-host abc123.us-east-1.aws.clickhouse.cloud \
  --sink-password "$CLICKHOUSE_PASSWORD"

filament source discover shop-pg

  Resource      Selectable     Primary key
  customers     ✓              id
  orders        ✓              id
  products      ✓              id

filament pipeline create pg-to-ch-snapshot \
  --source shop-pg \
  --sink clickhouse \
  --resources customers,products,orders \
  --sync-mode full \
  --write-mode replace \
  --sink-database shop

filament run pg-to-ch-snapshot

Run 106a899b-d1e3-5630-a671-c30390037bb6 completed: 1105000 records, 129873111 bytes.

CREATE TABLE shop.orders
(
    `id` Int64,
    `customer_id` Int64,
    `product_id` Int32,
    `quantity` Int32,
    `amount` Decimal(12, 2),
    `status` String,
    `ordered_at` DateTime64(6, 'UTC'),
    `updated_at` DateTime64(6, 'UTC'),
    `_filament_run_id` Nullable(String),
    `_filament_run_started_at` Nullable(DateTime64(6, 'UTC'))
)
ENGINE = MergeTree
ORDER BY id

filament pipeline create pg-to-ch-orders \
  --source shop-pg \
  --sink clickhouse \
  --resources orders \
  --sync-mode incremental \
  --write-mode upsert \
  --sink-database shop_live

filament run pg-to-ch-orders

UPDATE orders SET status = 'refunded', updated_at = now() WHERE id BETWEEN 1 AND 1000;
INSERT INTO orders SELECT 1000000 + g, 1 + g, 1 + g, 1, 49.99, 'placed', now(), now()
FROM generate_series(1, 500) g;

Run 28ab8805-ad4d-54ac-a46b-60af31eb1aa3 completed: 4237 records, 505506 bytes.

SELECT count() FROM shop_live.orders;
-- 1004237 physical rows before merges

SELECT count() AS current_rows,
       countIf(status = 'refunded' AND id <= 1000) AS refunded_updates
FROM shop_live.orders FINAL;
-- current_rows 1000500, refunded_updates 1000

-- PostgreSQL
SELECT status, count(*), sum(amount) FROM orders GROUP BY 1 ORDER BY 1;
-- ClickHouse
SELECT status, count(), sum(amount) FROM shop.orders GROUP BY status ORDER BY status;