go-pq-cdc is designed to provide efficient and lightweight Change Data Capture (CDC) for PostgreSQL databases. The architecture leverages PostgreSQL's built-in logical replication capabilities to capture changes in the database and stream these changes to downstream systems, such as Kafka, Elasticsearch etc. The entire system is written in Golang, ensuring low resource consumption and high performance.

Debezium vs go-pq-cdc benchmark

Capture existing data before starting CDC! The new snapshot feature enables initial data synchronization, ensuring downstream systems receive both historical and real-time data.

✨ Key Highlights:

• Zero Data Loss: Consistent point-in-time snapshot using PostgreSQL's pg_export_snapshot()
• Chunk-Based Processing: Memory-efficient processing of large tables
• Multi-Instance Support: Parallel processing across multiple instances
• Crash Recovery: Automatic resume from failures
• No Duplicates: Seamless transition from snapshot to CDC
• Snapshot Only Mode: One-time data export without CDC (no replication slot required)

📚 Read Full Documentation for detailed architecture, configuration, and best practices.

• go-pq-cdc
  • 📸 NEW: Snapshot Feature
    • Contents
    • Why?
    • Usage
    • Examples
    • Availability
    • TOAST Handling
    • Configuration
    • API
    • Exposed Metrics
    • Grafana Dashboard
    • Compatibility
    • Breaking Changes

CDC systems are crucial for real-time data synchronization, analytics, and event-driven architectures. Our main goal is to build a cdc base library for faster and stateful systems.

• Postgresql to Elasticsearch Connector
• Postgresql to Kafka Connector

go get github.com/Trendyol/go-pq-cdc

package main

import (
	"context"
	cdc "github.com/Trendyol/go-pq-cdc"
	"github.com/Trendyol/go-pq-cdc/config"
	"github.com/Trendyol/go-pq-cdc/pq/message/format"
	"github.com/Trendyol/go-pq-cdc/pq/publication"
	"github.com/Trendyol/go-pq-cdc/pq/replication"
	"github.com/Trendyol/go-pq-cdc/pq/slot"
	"log/slog"
	"os"
)

func main() {
	ctx := context.Background()
	cfg := config.Config{
		Host:      "127.0.0.1",
		Username:  "cdc_user",
		Password:  "cdc_pass",
		Database:  "cdc_db",
		DebugMode: false,
		Publication: publication.Config{
			CreateIfNotExists: true,
			Name: "cdc_publication",
			Operations: publication.Operations{
				publication.OperationInsert,
				publication.OperationDelete,
				publication.OperationTruncate,
				publication.OperationUpdate,
			},
			Tables: publication.Tables{publication.Table{
				Name:            "users",
				ReplicaIdentity: publication.ReplicaIdentityFull,
			}},
		},
		Slot: slot.Config{
			Name:                        "cdc_slot",
			CreateIfNotExists:           true,
			SlotActivityCheckerInterval: 3000,
		},
		Metric: config.MetricConfig{
			Port: 8081,
		},
		Logger: config.LoggerConfig{
			LogLevel: slog.LevelInfo,
		},
	}

	connector, err := cdc.NewConnector(ctx, cfg, Handler)
	if err != nil {
		slog.Error("new connector", "error", err)
		os.Exit(1)
	}

	defer connector.Close()
	connector.Start(ctx)
}

func Handler(ctx *replication.ListenerContext) {
	switch msg := ctx.Message.(type) {
	case *format.Insert:
		slog.Info("insert message received", "new", msg.Decoded)
	case *format.Delete:
		slog.Info("delete message received", "old", msg.OldDecoded)
	case *format.Update:
		slog.Info("update message received", "new", msg.NewDecoded, "old", msg.OldDecoded)
	}

	if err := ctx.Ack(); err != nil {
		slog.Error("ack", "error", err)
	}
}

• Simple
• Simple File Config
• Snapshot Mode (Initial Data Capture)
• Snapshot Only Mode (One-Time Export)
• PostgreSQL to Elasticsearch
• PostgreSQL to Kafka
• PostgreSQL to PostgreSQL

The go-pq-cdc operates in passive/active modes for PostgreSQL change data capture (CDC). Here's how it ensures availability:

• Active Mode: When the PostgreSQL replication slot (slot.name) is active, go-pq-cdc continuously monitors changes and streams them to downstream systems as configured.
• Passive Mode: If the PostgreSQL replication slot becomes inactive (detected via slot.slotActivityCheckerInterval), go-pq-cdc automatically captures the slot again and resumes data capturing. Other deployments also monitor slot activity, and when detected as inactive, they initiate data capturing.

This setup ensures continuous data synchronization and minimal downtime in capturing database changes.

PostgreSQL does not send TOASTed column values during an update unless the TOASTed column itself is modified. For example, if only a small field changes, large columns such as TEXT, JSONB, or BYTEA are sent as TOAST placeholders instead of real values.

To guarantee that UPDATE messages include full, correct data, go-pq-cdc automatically merges old and new tuples, restoring any missing TOASTed fields:

if m.OldTupleData != nil {
    for i, col := range m.NewTupleData.Columns {
        // Toasted columns are not sent unless the column itself changed
        if col.DataType == tuple.DataTypeToast {
            m.NewTupleData.Columns[i] = m.OldTupleData.Columns[i]
        }
    }
}

For this TOAST reconstruction to work, PostgreSQL must send the old tuple. This requires setting the table's replica identity to FULL:

ALTER TABLE your_table REPLICA IDENTITY FULL;

• With REPLICA IDENTITY DEFAULT, PostgreSQL only sends primary keys for updates/deletes.
• In that mode, old tuple values (including TOASTed fields) are not provided.
• Therefore, FULL is required to correctly handle TOASTed columns in CDC.

The client collects relevant metrics related to PostgreSQL change data capture (CDC) and makes them available at the /metrics endpoint.

Import the grafana dashboard json file.