DOCSEARCHbyAlgolia
Submit Docs

go-pq-cdc

repository·main·Indexed 41 days ago

https://github.com/trendyol/go-pq-cdc

A high-performance, lightweight Change Data Capture (CDC) tool written in Go that leverages PostgreSQL's logical replication to stream database changes to downstream systems like Kafka and Elasticsearch. It features a Snapshot Feature for initial data synchronization using pg_export_snapshot(), support for PostgreSQL 14+ streaming transactions, and a Visibility Guard to prevent race conditions in synchronous replication.

Tokens
24.1K
Snippets
51
Records
96
Agent score
47%

What's inside go-pq-cdc

  1. Overview of go-pq-cdc

    main

    go-pq-cdc is a lightweight and efficient Change Data Capture (CDC) tool for PostgreSQL databases. It is written in Go to ensure high performance and low resource consumption.

    Core Mechanism: It utilizes PostgreSQL's native logical replication capabilities to capture database changes and stream them to downstream systems like Kafka or Elasticsearch.

  2. Key features of the Snapshot mechanism

    main

    The Snapshot feature includes several capabilities designed for reliable and efficient data synchronization:

    • Consistent Point-in-Time Snapshot: Uses pg_export_snapshot() for transactional consistency.
    • Chunk-Based Processing: Splits large tables into manageable chunks to maintain memory efficiency.
    • Smart Partitioning: Automatically selects an optimal strategy (Integer Range, CTID Block, or Offset) based on the table structure.
    • Snapshot-Consistent Chunk Creation: All size queries (e.g., pg_relation_size) are executed within the snapshot transaction to ensure consistency.
    • Multi-Instance Support: Allows multiple instances to collaborate and process chunks in parallel.
    • Crash Recovery: Automatically resumes processing from the last successful point after a failure.
    • Zero Data Loss: Guarantees no gap between the snapshot completion and the start of the CDC phase.
    • Observability: Provides built-in metrics and progress tracking.

  3. Choose between Snapshot Modes: initial, never, and snapshot_only

    main

    The mode parameter determines how the connector handles historical data and its lifecycle.

    Featureinitial Modenever Modesnapshot_only Mode
    CDC after snapshot✅ Yes❌ No❌ No
    Replication slot✅ Required❌ Not created❌ Not created
    Publication✅ Required❌ Not created❌ Optional
    Process behaviorContinuousContinuousExits after snapshot
    Use caseCDC with bootstrapNo historical dataOne-time data export

    snapshot_only Mode Details

    • Multi-pod safe: Uses coordinator election to allow multiple pods to work together in parallel.
    • Resume capability: If a process is interrupted, restarting it will automatically continue from the last unfinished chunk.
    • No CDC: The process terminates once the snapshot is complete.
    snapshot:
  enabled: true
  mode: snapshot_only
  chunkSize: 10000

  4. Use Heartbeat-based WAL Protection

    main

    In low-traffic databases, replication slots can cause WAL bloat because the slot's restart_lsn lags behind, preventing WAL recycling. go-pq-cdc provides a table-based heartbeat mechanism to solve this.

    When heartbeat.table.name is configured, the connector:

    1. Creates the heartbeat table automatically.
    2. Ensures a single row exists.
    3. Periodically runs an internal UPDATE to generate small committed WAL records.
    4. Auto-acknowledges these events internally (they are not forwarded to your listener).

    Requirements for Heartbeat to work:

    1. The heartbeat table must be listed in publication.tables. If it is not part of the publication, the change is filtered out before reaching the slot, and WAL will not advance.
    2. The heartbeat table must have a REPLICA IDENTITY of DEFAULT (if it has a primary key) or FULL. Using REPLICA IDENTITY NOTHING will prevent the UPDATE from being decoded, causing the slot to stay frozen and WAL to grow.

    Best Practice: Always use a dedicated heartbeat table; never reuse a business table.

  5. How TOAST handling works in go-pq-cdc

    main

    PostgreSQL does not send TOASTed column values (large columns like TEXT, JSONB, or BYTEA) during an UPDATE unless the column itself is modified. This results in receiving only TOAST placeholders.

    To ensure UPDATE messages contain full, correct data, go-pq-cdc automatically merges old and new tuples, restoring missing TOASTed fields from the old tuple data.

  6. Understand Visibility Guard failMode behavior

    main

    The failMode setting determines how the stream behaves if the visibility guard cannot be reached or times out:

    • closed (default): If a timeout occurs or the guard connection encounters an error, the stream returns an error (new error class: "visibility guard unreachable") and restarts. This ensures data consistency by redelivering from the last confirmed LSN.
    • open: If a timeout occurs, the message is dispatched after the timeout with a warning logged and a counter incremented. This prioritizes availability over strict visibility guarantees.

    Note: There is no circuit breaker mechanism.

  7. How streamed transaction handling works in `protoVersion: 2`

    main

    In protoVersion: 2, PostgreSQL can split and interleave streamed transactions. To ensure data integrity and commit-safety, the stream sink buffers messages per transaction XID using the following logic:

    • STREAM START: Begins or continues buffering for the specific XID.
    • STREAM STOP: Keeps the buffered data but emits nothing.
    • STREAM COMMIT: Emits the buffered data for that XID and sets the last event WAL to the commit end LSN.
    • STREAM ABORT: Discards all buffered data for that XID.

    This mechanism ensures that transactions that are rolled back are never delivered to your CDC handlers.

  8. Understand the Visibility Guard concept

    main

    The Visibility Guard addresses a race condition in PostgreSQL synchronous replication. When a transaction commits, the WAL record is flushed and sent to go-pq-cdc before the transaction is marked as finished in the primary's snapshot. In synchronous replication, this gap can grow to a network round-trip, causing a consumer to receive an event for a row that is not yet visible to new snapshots (a MISS).

    The Visibility Guard allows the consumer to poll pg_current_snapshot() to certify that a transaction is visible before dispatching the event. This ensures that when a consumer receives an event, the data is guaranteed to be readable via a fresh snapshot on the primary.

  9. Understand the Snapshot Feature architecture

    main

    The Snapshot feature allows for consistent initial data loading (snapshots) followed by seamless transition to Change Data Capture (CDC). It uses a distributed architecture with a Coordinator and multiple Workers.

    Key Roles

    • Coordinator: Elected via PostgreSQL advisory locks. It manages the snapshot lifecycle, creates job metadata and chunks, and exports the initial snapshot transaction. It also acts as a worker.
    • Worker: Claims available chunks using SELECT FOR UPDATE SKIP LOCKED. Workers process chunk data within the exported snapshot transaction and update progress/heartbeats. All instances, including the coordinator, can act as workers.
    • Snapshotter: The main orchestrator managing the lifecycle.

    High-Level Flow

    1. Election: An instance becomes the Coordinator using pg_try_advisory_lock.
    2. Snapshot Export: The Coordinator captures the current LSN, starts a REPEATABLE READ transaction, and calls pg_export_snapshot() to get a snapshot_id. This transaction must remain open.
    3. Metadata Creation: The Coordinator uses the snapshot connection to query table sizes and create job/chunk metadata in the database.
    4. Parallel Processing: Workers (including the Coordinator) claim chunks and use SET TRANSACTION SNAPSHOT '<snapshot_id>' to read data consistent with the initial snapshot.
    5. CDC Transition: Once all chunks are processed, the snapshot transaction is closed, and CDC begins from the captured LSN, ensuring no data is lost or duplicated.

  10. Understand Chunk Processing and Heartbeats

    main

    Chunks are processed in parallel by multiple instances. To ensure reliability, the system implements the following:

    • Lock-Free Claiming: Workers use SELECT FOR UPDATE SKIP LOCKED to claim a chunk, preventing multiple workers from processing the same chunk without heavy contention.
    • Heartbeat Mechanism: Each worker runs a background goroutine that updates a heartbeat_at timestamp in the cdc_snapshot_chunks table every 10 seconds.
    • Stale Chunk Reclamation: If a worker crashes or freezes, its heartbeat will stop updating. Other workers can identify these stale chunks (where heartbeat_at < NOW() - 30s) and reclaim them to ensure the job eventually completes.

  11. How the Snapshot Feature lifecycle works

    main

    The snapshot feature ensures data consistency when performing an initial load before starting CDC (Change Data Capture). It follows five distinct phases:

    1. Coordinator Election: Multiple instances use PostgreSQL advisory locks (pg_try_advisory_lock) to elect a single COORDINATOR. Other instances become WORKERS. The coordinator is responsible for capturing the LSN, exporting the snapshot, and splitting tables into chunks.
    2. Snapshot Export: The coordinator starts a REPEATABLE READ transaction, calls pg_export_snapshot(), and keeps the transaction open. This exported snapshot ID is used by workers to see the exact same data state.
    3. Metadata Creation: The coordinator creates job and chunk metadata. Crucially, it queries table sizes (pg_relation_size) using the same connection that holds the open snapshot transaction to ensure the chunking boundaries are consistent with the snapshot.
    4. Chunk Processing: Both the coordinator and workers process chunks in parallel. They use SELECT FOR UPDATE SKIP LOCKED to claim chunks atomically. Each chunk is processed within its own transaction that uses SET TRANSACTION SNAPSHOT '<snapshot_id>' to maintain consistency.
    5. CDC Continuation: Once all chunks are processed, the coordinator marks the job as complete, commits the export transaction, and starts the logical replication stream from the LSN captured in Phase 2. This ensures no data overlap or gaps between the snapshot and the CDC stream.

  12. Configure Replica Identity for TOAST Handling

    main

    To correctly reconstruct TOASTed columns (large values stored out-of-line) during CDC, PostgreSQL must be configured to send the old tuple data. This is controlled by the REPLICA IDENTITY setting on your tables.

    Options for Replica Identity:

    1. REPLICA IDENTITY FULL (Recommended for full TOAST reconstruction):

      • Sends the old values of all columns.
      • Required if you need to access old values of TOASTed columns during updates/deletes.
      • Command: ALTER TABLE your_table REPLICA IDENTITY FULL;

    2. REPLICA IDENTITY USING INDEX:

      • Sends only the old values of the columns included in a specific unique index.
      • Reduces network payload size compared to FULL.
      • Limitation: Old values for columns not in the index are not available.
      • Command: ALTER TABLE your_table REPLICA IDENTITY USING INDEX your_index_name;

    3. REPLICA IDENTITY DEFAULT:

      • Only sends primary keys for updates/deletes.
      • Limitation: Old tuple values (including TOASTed fields) are not provided. This is insufficient for full TOAST reconstruction.

    4. REPLICA IDENTITY NOTHING:

      • PostgreSQL does not provide old-row data for updates and deletes.
      • Best used for insert-only publications.
    ALTER TABLE your_table REPLICA IDENTITY FULL;