postgres

package
v0.4.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 30, 2026 License: Apache-2.0 Imports: 39 Imported by: 0

Documentation

Overview

Package source.postgres implements the PostgreSQL replication source: a single logical-decoding reader (pgoutput) that decodes row changes into rowchange.Change, positions them at their commit LSN, and exposes the synced and confirmed positions for the DBLog watermark logic. The snapshot path reuses the source-agnostic DBLog orchestrator.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func AdvanceSlot added in v0.3.0 

func AdvanceSlot(ctx context.Context, db *sql.DB, slotName string, target position.LSN) error

AdvanceSlot advances the logical slot's confirmed_flush_lsn to target, letting the server recycle WAL the sink has already committed (#156). It is idempotent and must run while the slot is inactive — before START_REPLICATION attaches the walsender.

func CanonicalSchema 

func CanonicalSchema(tbl *TableState) (core.Schema, error)

CanonicalSchema derives the canonical core.Schema from the introspected table. Mappable columns map to their canonical kind; unmappable ones are carried as KindUnknown so a declared cast can land them explicitly — never silent coercion. The source knows nothing about any sink.

func ConfirmedLSN 

func ConfirmedLSN(ctx context.Context, db *sql.DB, slotName string) (*position.LSN, error)

ConfirmedLSN reads the slot's confirmed_flush_lsn: the stream start for a first boot (no committed resume position).

func EnsureSetup 

func EnsureSetup(ctx context.Context, db *sql.DB, slotName string, tables []source.TableRef, plugin string) error

EnsureSetup makes the server side ready, idempotently:

  • REPLICA IDENTITY FULL on every replicated table, so updates and deletes carry the full old row (parity with MySQL's row_image=FULL);
  • for pgoutput, the logical publication listing exactly the pipeline's tables;
  • the logical decoding slot with the selected plugin.

The slot is the consistency anchor: created before the snapshot starts, it guarantees no transaction between slot creation and the stream start is ever lost.

func ValidateSlotState added in v0.3.0 

func ValidateSlotState(ctx context.Context, db *sql.DB, slotName string, resume *position.LSN, logger *slog.Logger) (*position.LSN, error)

ValidateSlotState reconciles a stored resume LSN with the slot's confirmed_flush_lsn and returns the LSN the stream should start from (#156):

  • resume == nil (first boot): start from the slot's confirmed point.
  • resume == confirmed: no drift.
  • resume < confirmed: the slot is ahead of the sink. The sink's stored point is behind what the server already confirmed (and may have recycled), so the slot's point is authoritative — warn and use it.
  • resume > confirmed: the sink durably committed past the slot's last confirmed_flush_lsn (the server learns the applied position from a later standby-status update, so a crash can leave the slot behind). This is recoverable: the WAL from resume onward is still retained because the slot never confirmed past it, so advance the slot to the sink's point rather than forcing a re-snapshot.

Types

type Chunker 

type Chunker struct {
	// contains filtered or unexported fields
}

Chunker splits a table into snapshot chunks. The strategy is CTID by default; a configured chunk column switches to a key-based strategy (batch-size for integer/float, next-query otherwise). A partitioned table (workers > 1) always uses a key-based strategy, because the chunk range must stay routable to the same worker as the live stream.

func NewChunker 

func NewChunker(ctx context.Context, db *sql.DB, source, pk string, chunkSize int, opts ...ChunkerOption) (*Chunker, error)

NewChunker builds a chunker for one source table.

func (*Chunker) Bounds 

func (c *Chunker) Bounds(ctx context.Context) ([][]any, error)

Bounds returns the ordered chunk boundary keys for the resolved strategy. The whole computation is retried on a transient error.

func (*Chunker) PK 

func (c *Chunker) PK() []string

PK returns the primary key columns.

func (*Chunker) Partitions added in v0.3.0 

func (c *Chunker) Partitions(ctx context.Context, n int) ([]source.Chunk, error)

Partitions divides this table's single-column primary key into n contiguous, ordered ranges for worker partitioning — the same ranges govern the DBLog snapshot and the live-stream routing, so a key can never switch partition ownership between the two phases (no hash is involved, so the two phases cannot disagree).

It always returns exactly n disjoint ranges (an empty or sparse table is padded with degenerate or type-domain ranges), and marks this chunker key-partitioned: CTID ranges are physical and cannot be clipped to a PK range, so a partitioned table must chunk by key.

func (*Chunker) Scan 

func (c *Chunker) Scan(ctx context.Context, ch source.Chunk, fn func(row map[string]any) error) error

Scan executes the chunk SELECT and calls fn for every row, keyed by column name. Values decode through the same scalar mapping the pgoutput reader uses, so snapshot rows and stream rows land in Iceberg identically.

The whole scan is retried on a transient error, covering a failure during rows.Next/Scan and not just the initial QueryContext. Rows are buffered until the scan completes, so a retry never re-delivers rows to fn.

Within one scan, rows are normalized by a bounded pool of `workers` goroutines, but fn is always invoked from a single goroutine, so it need not be thread-safe. Row order is not preserved (snapshot rows are keyed inserts, order-independent).

type ChunkerOption added in v0.3.0 

type ChunkerOption func(*Chunker)

ChunkerOption tunes a Chunker.

func WithChunkColumn added in v0.3.0 

func WithChunkColumn(col string) ChunkerOption

WithChunkColumn selects the chunk column (#151). Empty keeps the CTID default.

func WithColumns added in v0.3.0 

func WithColumns(cols []string) ChunkerOption

WithColumns sets the column projection (#162). Empty means all columns.

func WithFilter added in v0.3.0 

func WithFilter(f sq.Sqlizer) ChunkerOption

WithFilter sets the compiled structured filter (#163). Nil means none.

func WithRetries added in v0.3.0 

func WithRetries(n int) ChunkerOption

WithRetries sets the transient-error retry budget for snapshot queries.

func WithTargetBytes added in v0.3.0 

func WithTargetBytes(n int64) ChunkerOption

WithTargetBytes sets the CTID strategy's target bytes per chunk (from sink.defaults.targetFileSize). Non-positive keeps the default.

func WithWorkers added in v0.3.0 

func WithWorkers(n int) ChunkerOption

WithWorkers sets the concurrent normalization pool size. n < 1 is clamped to 1 (sequential).

type Column 

type Column struct {
	Name     string
	DataType string // pg_type native name ("int8", "text", "timestamptz", …)
	NotNull  bool   // attnotnull: the column cannot hold NULL
}

Column is one introspected source column.

type Config 

type Config struct {
	URI      string
	ConnCfg  *ConnConfig // resolved from nested config (nil when using URI)
	DB       *sql.DB
	SlotName string
	Tables   []source.TableRef
	Logger   *slog.Logger
	// RetryCount is the transient-error reconnect budget for the
	// replication stream (#166). 0 disables reconnect.
	RetryCount int
	// InitialWait bounds how long the reader waits for the first WAL message
	// before failing with a non-retryable error (#154). Zero uses the
	// default (300s).
	InitialWait time.Duration
	// Plugin is the logical decoding plugin: "pgoutput" (default) or
	// "wal2json".
	Plugin string
	// Filters and Columns are the per-source read projection (#162/#163),
	// keyed by "schema.table". The reader compiles the filter with the
	// introspected column types, so numeric columns compare numerically.
	Filters map[string]*spec.Filter
	Columns map[string][]string
}

Config dials one PostgreSQL instance. One replication connection per source — the invariant the whole design stands on. DB is the query connection used for introspection and position reads.

type ConnConfig added in v0.3.0 

type ConnConfig struct {
	// QueryURI is the libpq DSN rendered from the config. It is what travels
	// to a distributed worker (the worker only receives kind + dsn).
	QueryURI string
	// ConnConfig is the native pgx config for replication and query
	// connections. It carries TLSConfig (built by pgx from the DSN) and
	// DialFunc (SSH) so both connection paths share one transport.
	ConnConfig *pgx.ConnConfig
	// MaxOpenConns is the resolved maxThreads value.
	MaxOpenConns int
	// RetryCount is the resolved number of transient-connection retries.
	RetryCount int
	// InitialWaitTime is the resolved CDC initial-WAL-message wait.
	InitialWaitTime time.Duration
	// Plugin is the resolved logical decoding plugin ("pgoutput"|"wal2json").
	Plugin string
	// contains filtered or unexported fields
}

ConnConfig holds the resolved connection parameters, built from either a URI or the nested PostgresSource config.

func BuildConnConfig added in v0.3.0 

func BuildConnConfig(uri string) (*ConnConfig, error)

BuildConnConfig constructs a ConnConfig from a URI string.

func BuildConnConfigFromPostgres added in v0.3.0 

func BuildConnConfigFromPostgres(pg *spec.PostgresSource) (*ConnConfig, error)

BuildConnConfigFromPostgres constructs a ConnConfig from the nested PostgresSource config. When present, source.uri is ignored for connection building.

The config is created by pgx.ParseConfig on the rendered DSN — never by hand. pgx's ConnectConfig panics on a config it did not create, and it also needs RuntimeParams initialized (Reader.New writes "replication" into it). Parsing the DSN also gets pgx's native TLS handling for free (require, verify-ca and verify-full all follow libpq semantics, including loading the CA and client cert/key). Only the SSH DialFunc is layered on afterwards.

func (*ConnConfig) Close added in v0.3.0 

func (c *ConnConfig) Close() error

Close tears down the SSH tunnel, if one was established. Safe to call multiple times.

type Projection added in v0.3.0 

type Projection struct {
	Columns []string
	// contains filtered or unexported fields
}

Projection is a table's source-side read projection: the columns to emit and the compiled filter a row must satisfy. Both are applied on the decoded map, before the Arrow hot-path.

type Reader 

type Reader struct {
	// contains filtered or unexported fields
}

Reader wraps one logical-decoding connection and decodes pgoutput row changes into rowchange.Change, positioned at their commit LSN.

func New 

func New(ctx context.Context, cfg Config, out chan<- rowchange.Change) (*Reader, error)

New introspects the tables, performs the server-side setup (replica identity, publication, slot), and opens the replication connection — but does not start streaming.

func (*Reader) ClearWindow 

func (r *Reader) ClearWindow()

ClearWindow closes the DBLog window opened by OpenWindow.

func (*Reader) Close 

func (r *Reader) Close()

Close stops the reader and its replication connection. It cancels the replication loop first and waits for it to exit pgx, so the conn close never races an in-flight receive.

func (*Reader) Master 

func (r *Reader) Master(ctx context.Context) (position.Position, error)

Master reports the server's current WAL write position (the caught-up target for the DBLog window).

func (*Reader) OpenWindow 

func (r *Reader) OpenWindow(ctx context.Context, chunkID uint32)

OpenWindow opens the DBLog window for chunkID. The reader captures its source watermark — the server's current WAL position — and, from now on, tags decoded events whose commit LSN is strictly past that watermark InWindow for the chunk, until ClearWindow. Transactions at or before the watermark are already reflected in the chunk SELECT (or an earlier chunk) and must not be tagged. The tag is applied synchronously at decode — no event can escape the window by racing a channel pull. The watermark is an LSN (64-bit, monotonic), never the 32-bit transaction xid — the pgoutput Begin message only carries the raw xid, which wraps every ~4 billion transactions; the LSN does not.

func (*Reader) SetConfirmed 

func (r *Reader) SetConfirmed(f func() position.Position)

SetConfirmed installs the confirmed-position callback.

func (*Reader) StartFromLSN 

func (r *Reader) StartFromLSN(ctx context.Context, at *position.LSN) error

StartFromLSN begins streaming from the given LSN, blocking until the stream ends or ctx is cancelled. Call in a goroutine. An LSN of 0/0 starts at the slot's confirmed point.

A transient stream failure (network blip, 08xxx/53xxx, admin shutdown) is retried: the replication connection is re-dialed and the stream resumes from the last position committed to the sink, up to r.retries times (#166). A permanent failure is returned immediately.

func (*Reader) Synced 

func (r *Reader) Synced() position.Position

Synced reports the end LSN of the last committed transaction.

type Source 

type Source struct {
	// contains filtered or unexported fields
}

Source adapts the Postgres pgoutput reader to the source contract. It is the self-contained driver entry point; init() registers the kind.

func (Source) CloseQuery 

func (a Source) CloseQuery() error

CloseQuery releases the query connection and tears down the SSH tunnel.

func (Source) Discover added in v0.3.0 

func (a Source) Discover(ctx context.Context, schemas []string) ([]source.TableRef, error)

Discover implements source.Discoverer (#152): every table the connected user may SELECT, limited to schemas when non-empty. It returns base tables

  • matviews ('m') and foreign tables ('f') are excluded because EnsureSetup runs REPLICA IDENTITY FULL and CREATE PUBLICATION on every discovered table, and both fail on those relkinds;
  • leaf partitions are excluded (relispartition) because the parent already covers their rows — replicating parent and leaves doubles every row, and syncPublication's pg_publication_rel model expects the parent.

func (Source) Incremental added in v0.3.0 

func (a Source) Incremental(ctx context.Context, t source.TableRef, cursor, after string) (string, []map[string]any, error)

Incremental implements source.IncrementalSource (#157): one cursor pass, SELECT <projection> FROM <table> WHERE [filter AND] <cursor> > $1 ORDER BY <cursor>. It returns the new cursor — the last row's cursor value, or "" when no rows — and the decoded rows. No slot, no publication.

func (Source) InitialPosition 

func (a Source) InitialPosition(ctx context.Context) (position.Position, error)

InitialPosition returns the slot's confirmed LSN: a first boot starts streaming from the slot's consistency point — everything after it flows through the stream, everything before it belongs to the snapshot.

func (Source) Introspect 

func (a Source) Introspect(ctx context.Context, t spec.Table) (core.TableRef, core.Schema, []core.Warning, error)

Introspect resolves one spec table into its ref and canonical schema.

func (Source) NewChunker 

func (a Source) NewChunker(source, pk string, chunkSize int) (source.ChunkSource, error)

NewChunker builds the chunk SELECT source for one table. The concurrent normalization pool (#161) and the snapshot-query retry budget (#166) come from the resolved connection config; the column projection (#162), the filter (#163), and the chunk column / CTID target bytes (#151) come from the spec.

func (Source) Open 

func (a Source) Open(ctx context.Context, refs []source.TableRef) (source.Reader, error)

Open builds the replication reader over the pipeline's tables.

func (Source) ParsePosition 

func (a Source) ParsePosition(s string) (position.Position, error)

ParsePosition decodes a stored cdc.position as an LSN.

type TableState 

type TableState struct {
	Schema    string
	Name      string
	Columns   []Column
	PKColumns []int
}

TableState is the introspection result for one source table: the ordered column list and the primary key column indexes, so row decoding and Iceberg schema derivation share one path.

func QueryTable 

func QueryTable(ctx context.Context, db *sql.DB, schemaName, tableName string) (*TableState, error)

QueryTable introspects one table via pg_catalog, producing the column order the pgoutput relation messages follow.

func (*TableState) FindColumn 

func (t *TableState) FindColumn(name string) int

FindColumn returns the column index by name, or -1.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.