postgres

package
v0.5.0 Latest Latest
Warning

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

 Go to latest Published: Oct 11, 2026 License: Apache-2.0 Imports: 44 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, replicaIdentity 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) 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).

func (*Chunker) ScanArrow added in v0.5.0 

func (c *Chunker) ScanArrow(ctx context.Context, ch source.Chunk, enc *transport.RowEncoder, expected int) (int, error)

ScanArrow reads one chunk straight into enc, one snapshot insert per row, and returns the row count; expected is the chunk's expected row count. An encoder the caller sized already (ReservePerRow) is used as is; otherwise it is sized from the chunk's first rows.

Scan hands every row over as a map of Go values, which the caller then encodes: the chunk was held as maps and again as Arrow, the most memory-expensive path in the Postgres snapshot (#590). Here a text, JSON or binary cell goes from the driver's buffer into the column's Arrow data buffer, and the buffers are sized once for the chunk. Any other cell is converted exactly as Scan converts it.

The read runs in one REPEATABLE READ, READ ONLY transaction, as Scan does, so the chunk sees a single consistent snapshot. Transaction and query setup are retried before the first row is appended; mid-scan failures are not retried, since retrying after rows have been appended would double-count them (they fall back to the coordinator's chunk redo on worker loss).

func (*Chunker) ScanArrowPages added in v0.5.0 

func (c *Chunker) ScanArrowPages(ctx context.Context, ch source.Chunk, enc *transport.RowEncoder, maxBytes int, emit func(rec arrow.RecordBatch, n int) error) (int, error)

ScanArrowPages reads one chunk straight into enc, cutting it into pages when the accumulated data bytes reach maxBytes. emit is called per page with the page's record (the caller owns it) and its row count; it may block (the worker's backpressure). The read runs in one REPEATABLE READ, READ ONLY transaction — the same as ScanArrow — so every page sees one consistent snapshot; the transaction is committed after the last page (#622).

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 is the SQL-standard BASE type WITHOUT a modifier
	// ("character varying", "numeric", "timestamp with time zone") — the form
	// every switch compares against and every `?::<type>` cast accepts.
	DataType string
	// RawType is format_type with the modifier ("character varying(64)",
	// "numeric(10,2)"), kept only so decimal precision/scale can be parsed
	// (issue #561).
	RawType string
	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
	// ReplicaIdentity is "full" (default) or "keep" (issue #571): FULL is set
	// only when the table is not already, and keep leaves the table's
	// identity untouched.
	ReplicaIdentity 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
	// UpsertTargets marks the target tables that maintain keyed state. An
	// UPDATE that changes the primary key emits a delete of the old key only
	// for these; append targets keep the old row.
	UpsertTargets map[string]bool
	// Schemas is each target table's canonical (projected) schema, the shape
	// the direct encoder builds. The engine's SetSourceSchemas replaces
	// KindUnknown with the cast-resolved type before Start (#455).
	Schemas map[string]core.Schema
	// OnTruncate is "ignore" (default) or "fail": whether a TRUNCATE on a
	// replicated table ends the stream (#671).
	OnTruncate string
	// OnDestructiveDDL, when non-nil, is called for every TRUNCATE seen on the
	// stream so the engine can record a metric and a run event (#671).
	OnDestructiveDDL func(source.DestructiveDDL)
}

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 {
	// contains filtered or unexported fields
}

Projection is a table's source-side read filter: the compiled filter a row must satisfy. The emitted column set is the canonical schema (the adapter filters it), so the live path appends the projected columns straight into Arrow; only the filter needs named access, and it reads a minimal map of just the columns it references.

type Reader 

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

Reader wraps one logical-decoding connection and decodes pgoutput/wal2json row changes straight into Arrow wire batches (emitted on batchOut at commit), positioned at their commit LSN.

func New 

func New(ctx context.Context, cfg Config, batchOut chan<- *dataplane.Batch) (*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 is part of the SourceReader contract; see 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(_ context.Context, _ uint32)

OpenWindow is part of the SourceReader contract. The DBLog window tag is applied by the coordinator's gate to the batches it holds (WindowTag on the wire) — not by the source — so the reader keeps no window state.

func (*Reader) SetConfirmed 

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

SetConfirmed installs the confirmed-position callback.

func (*Reader) SetResume added in v0.5.0 

func (r *Reader) SetResume(pos string)

SetResume records the position the stream resumes from: the safe position a partial piece of the first transaction carries.

func (*Reader) SetSourceSchemas added in v0.5.0 

func (r *Reader) SetSourceSchemas(resolved map[string]core.Schema)

SetSourceSchemas installs the cast-resolved canonical schemas before Start. The direct encoder must see the resolved kinds.

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, bool, error)

func (Source) IncrementalBatch added in v0.5.0 

func (a Source) IncrementalBatch(ctx context.Context, t source.TableRef, cursor, after string, schema core.Schema) (string, *dataplane.Batch, bool, error)

IncrementalBatch implements source.BatchIncrementalSource (#733): the columnar sibling of Incremental. It runs the SAME cursor-bounded page query (incrementalQuery), but encodes every row straight into the Arrow builders of a transport.RowEncoder built from schema, with no per-row map[string]any in between. schema is the engine's canonical wire schema for the target table (the shape it installs on the worker), so the returned batch lands exactly what the map path's RecordFromChanges would have produced.

The returned batch carries the page's rows, every one stamped with the page's resume position (as the map path stamps every change with next), and Watermark = []byte(next). A page with no rows returns (next, nil, false, nil), which the engine treats as the end of the drain.

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.