Streaming Query Replication of PostgreSQL To Snowflake

DROP TABLE IF EXISTS pgbench_history;

CREATE TABLE pgbench_history (
    tid integer,
    bid integer,
    aid integer,
    delta integer,
    mtime timestamp without time zone,
    filler character(22)
);

INSERT INTO pgbench_history (tid, bid, aid, delta, mtime, filler)
SELECT
    (random() * 10)::int,
    (random() * 10)::int,
    (random() * 100000)::int,
    (random() * 10000 - 5000)::int,
    now() - (random() * interval '30 days'),
    lpad('', 22)
FROM generate_series(1, 10000);

CREATE INDEX pgbench_history_mtime_idx ON pgbench_history(mtime);

CREATE TABLE pgbench_history (tid integer, bid integer,
aid integer, delta integer, mtime timestampntz, filler text);

CREATE MIRROR postgres_to_snowflake_tutorial
FROM postgres_peer TO snowflake_peer FOR
$$SELECT * FROM public.pgbench_history WHERE mtime BETWEEN {{.start}} AND {{.end}}$$
WITH (
	watermark_column = 'mtime',
	watermark_table_name = 'pgbench_history',
	mode = 'append',
	parallelism = 8,
	refresh_interval = 10,
	num_rows_per_partition = 50000,
	destination_table_name = 'public.pgbench_history'
);

SELECT count(*) FROM postgres_peer.pgbench_history;
SELECT count(*) FROM snowflake_peer.public.pgbench_history;