README
¶
go-pq-cdc-pq
PostgreSQL Change Data Capture (CDC) library for Go.
Installation
go get github.com/Trendyol/go-pq-cdc-pq
Usage
See examples in the example/ directory. Snapshot bootstrapping from go-pq-cdc v1.0.3+ is supported via the cdc.snapshot block in your config. For advanced options (mode, chunkSize, etc.) check the upstream snapshot feature guide.
Table primary key mapping
By default, upsert/delete queries use primary key column name id. If you have multiple tables with different primary key names, you can provide a per-table mapping via config (tablePrimaryKeys) or via options (WithTablePrimaryKeys).
Config example:
tablePrimaryKeys:
users: id
mock_data_table: id_pk
Option example:
gopqcdcpq.WithTablePrimaryKeys(map[string]string{
"users": "id",
"mock_data_table": "id_pk",
})
License
MIT
Documentation
¶
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Config ¶
type Config struct {
ConfigPath string
PublicationTables []publication.Table
PrimaryKey string
DefaultSchema string
TablePrimaryKeys map[string]string
Mapper Mapper
}
Config captures user configurable settings for the connector.
type Connector ¶
type HandlerFunc ¶
HandlerFunc adapts a function to the Handler interface.
type Mapper ¶ added in v0.0.4
type Mapper func(event *Message) []QueryAction
Mapper transforms CDC events into SQL query actions. It receives a Message containing the CDC event data and returns a slice of QueryAction to be executed on the target database. Returning nil or empty slice will skip the event (ack will still be called).
type Message ¶
type Message struct {
EventTime time.Time
LSN string // WAL LSN in Postgres "X/Y" format; "" if unavailable
TableName string
TableNamespace string
OldData map[string]any
NewData map[string]any
Type MessageType
// Internal fields for batch processing
Query string
Args []any
Ack func() error
Schema string
Table string
Action string // INSERT, UPDATE, DELETE - kept for backward compatibility
OldKeys map[string]any
NewValues map[string]any
}
func NewDeleteMessage ¶
func NewInsertMessage ¶
func NewSnapshotMessage ¶
func NewUpdateMessage ¶
type MessageType ¶
type MessageType string
const ( InsertMessage MessageType = "INSERT" UpdateMessage MessageType = "UPDATE" DeleteMessage MessageType = "DELETE" SnapshotMessage MessageType = "SNAPSHOT" )
func (MessageType) IsDelete ¶
func (m MessageType) IsDelete() bool
func (MessageType) IsInsert ¶
func (m MessageType) IsInsert() bool
func (MessageType) IsSnapshot ¶
func (m MessageType) IsSnapshot() bool
func (MessageType) IsUpdate ¶
func (m MessageType) IsUpdate() bool
type Option ¶
type Option func(*Config)
Option represents an option for customizing the connector config.
func WithConfigPath ¶
WithConfigPath overrides the location of the main configuration file.
func WithDefaultSchema ¶
WithDefaultSchema overrides the default schema name used for message processing. Defaults to "public".
func WithMapper ¶ added in v0.0.4
WithMapper overrides the default mapper function used to transform CDC events into SQL queries. This allows custom transformation logic such as: - Writing to different tables - Transforming field names or values - Filtering events - Writing to multiple tables from a single event
func WithPrimaryKey ¶
WithPrimaryKey overrides the default primary key column name used for building upsert and delete queries. Defaults to "id".
func WithPublicationTables ¶
func WithPublicationTables(tables ...publication.Table) Option
WithPublicationTables overrides the publication tables used while creating the CDC publication. When not provided, defaults are used.
func WithTablePrimaryKeys ¶ added in v0.0.2
WithTablePrimaryKeys overrides the default primary key per table. Keys are table names (for example: "fake_content_analysis") and values are the corresponding primary key column names. When not provided for a table, the default PrimaryKey is used.
type QueryAction ¶ added in v0.0.4
QueryAction represents a SQL query to be executed on the target database.
func DefaultMapper ¶ added in v0.0.4
func DefaultMapper(event *Message) []QueryAction
DefaultMapper performs 1:1 replication using the same schema. It simply forwards the pre-built query from the Message.
type Sink ¶
type Sink struct {
// contains filtered or unexported fields
}
Sink applies CDC messages to the target PostgreSQL database. Batches messages and flushes when BulkSize is reached or Timeout expires. Implements retry logic with exponential backoff. Messages are acknowledged only after successful execution.
func NewSink ¶
NewSink creates a new Sink instance.
Example:
pool, _ := database.NewTargetPool(ctx, cfg) sink := NewSink(pool, batchConfig, slog.Default())
func (*Sink) Start ¶
Start consumes messages from the channel and applies them to the target database. Runs until context is cancelled. Flushes when BulkSize is reached or Timeout expires. On shutdown, flushes any remaining buffered messages.
Example:
messages := make(chan Message, 1000) go sink.Start(ctx, messages)
Source Files
Directories