The First Hurdle: Snapshotting

A CDC pipeline is designed to stream ongoing changes, but it must first be initialized with the data that already exists in the source tables. This process of performing an initial, full copy of the data is known as snapshotting or the initial load.

Why snapshots break

The central problem is consistency. Without a boundary, writes that land mid-run show up twice (snapshot + stream) or disappear entirely. Retention clocks keep ticking and, if the log purges, the only fix is to start over.

Picture a payments table: you start scanning, finance posts refunds, and analytics reconciles against a different truth hours later. The fallout is expensive clean-up, tense incident calls, and a loss of trust in the pipeline.

A staged handoff

Modern log-based CDC tools coordinate a point-in-time snapshot with the transaction log so the handoff to streaming has no gaps or double-counts. Treat the run in three phases:

Copy-- Requirements (once/admin):
ALTER SYSTEM SET wal_level = logical;
ALTER SYSTEM SET max_replication_slots = 10;  -- adjust
SELECT pg_reload_conf();

-- Boundary + consistent read (within one tx while snapshotting):
BEGIN TRANSACTION ISOLATION LEVEL REPEATABLE READ;
SELECT pg_current_wal_lsn();  -- record this
-- Keep transaction open during table scans; Debezium coordinates via logical slot.

Copy-- Requirements:
-- binlog_format=ROW; binlog_row_image=FULL; GTID (preferred)

-- Boundary + consistent snapshot:
SHOW MASTER STATUS;                  -- or SELECT @@global.gtid_executed;
START TRANSACTION WITH CONSISTENT SNAPSHOT;
-- Read tables in PK order; Debezium avoids FTWRL in OLTP paths.

Copy-- Requirements (once/admin):
ALTER DATABASE YourDb SET ALLOW_SNAPSHOT_ISOLATION ON;
ALTER DATABASE YourDb SET READ_COMMITTED_SNAPSHOT ON;
EXEC sys.sp_cdc_enable_db;  -- if using native CDC

-- Boundary + snapshot:
SELECT sys.fn_cdc_get_max_lsn();     -- record
SET TRANSACTION ISOLATION LEVEL SNAPSHOT;
BEGIN TRAN;
-- Read tables; keep tx open during snapshot.

Copy-- Requirements (once/admin):
ALTER DATABASE ADD SUPPLEMENTAL LOG DATA;  -- minimal or ALL columns for PK-less

-- Boundary:
SELECT CURRENT_SCN FROM V$DATABASE;        -- record
-- Consistent reads use UNDO/flashback; ensure UNDO retention covers snapshot.

Snapshot modes (choose intentionally)

• initial: Do a snapshot then continue streaming (default for first run).
• initial_only: One-time bulk load, no streaming (bootstraps batch targets).
• schema_only / never: Skip data snapshot (for pre-seeded targets or special cases).
• incremental snapshot: Chunk tables online (no global pause) using low/high watermarks and primary-key ranges; supports resuming mid-table.

Performance: chunking & throughput

• Chunking: Read large tables in PK-ordered chunks (5k–200k rows) to bound memory and reduce long transactions. Prefer monotonic PKs; otherwise use synthetic chunking keys.
• Parallelism: Limit concurrent table snapshots to protect OLTP (1–3 in parallel).
• Throttling: Use fetch size / max rows per poll to keep steady pressure, not bursts.
• Replica reads: For MySQL/PG, consider reading from a replica to offload the primary (ensure replication lag is acceptable and log positions still align).

Operational guardrails (day-one checks)

• Log retention: Ensure WAL/binlog/T-log won’t be purged before snapshot finishes. Alert on backlog age and size.
• Privileges: Replication/log-read permissions; publication/slot creation where applicable.
• DDL during snapshot: Prefer compatibility rules (schema registry) and deploy windows; avoid table rebuilds that invalidate consistent reads.
• Sink idempotency: Upsert/delete semantics in the warehouse/search index must be idempotent so replays (post-snapshot) don’t duplicate data.
• Resume safely: Snapshots should be restartable: resume from last completed chunk and the recorded high-watermark.

CopyMERGE INTO dw.customers AS t
USING staging.customers AS s
ON t.customer_id = s.customer_id
WHEN MATCHED THEN UPDATE SET
  name = s.name,
  email = s.email,
  updated_at = s.updated_at
WHEN NOT MATCHED THEN INSERT (customer_id, name, email, updated_at)
VALUES (s.customer_id, s.name, s.email, s.updated_at);

CopySELECT id, COUNT(*) AS c
FROM dw.orders
GROUP BY id
HAVING COUNT(*) > 1;