source

type BatchIncrementalSource ¶ added in v0.5.0

type BatchIncrementalSource interface {
	IncrementalBatch(ctx context.Context, t TableRef, cursor string, after string, schema core.Schema) (next string, batch *dataplane.Batch, more bool, err error)
}

BatchIncrementalSource is the optional COLUMNAR sibling of IncrementalSource (#733). It runs the same cursor-bounded page read but encodes the page straight into a wire-schema Arrow batch, so no per-row map[string]any exists between the driver and the data plane.

schema is the target table's canonical WIRE schema — the same shape the engine installs on the worker (SetKnownSchema) and would encode the map path against — so the returned batch is byte-for-byte what RecordFromChanges would have produced from Incremental's rows. Without it the source would have to infer a schema, which drifts from the engine's (order, types, enrichment columns); the interface deliberately carries it so the two paths cannot diverge.

The cursor/tie-break semantics are identical to IncrementalSource: next is the cursor to resume from, more reports whether another page is already available, and the returned batch's Watermark is []byte(next). A source may implement this and/or IncrementalSource; the engine prefers this one when present, because it skips the row maps entirely, and falls back to IncrementalSource otherwise.

A source that returns an empty page returns (next, nil, false, nil); the engine treats a nil batch as the end of the drain.

type Capabilities ¶

type Capabilities struct {
	Snapshot   bool
	ChunkQuery bool
	Stream     bool
	// MaxConnections is the upper bound on concurrent query connections
	// the source tolerates. 0 means no opinion (unlimited).
	MaxConnections int
	// Modes lists the synchronization modes this source supports.
	Modes []Mode
	// BeforeImage reports whether the source carries the row image that was
	// deleted on a DELETE. MySQL (binlog row) always does; Postgres only
	// per replica identity; Kafka does not (tombstones are null). Append
	// tables with onDelete: record depend on this.
	BeforeImage bool
	// MonotonicSequence reports whether the transport coordinate of a
	// message never reappears (Kafka offset, Kinesis sequence) — the
	// property append-idempotent relies on.
	MonotonicSequence bool
	// RecoverablePosition reports whether the source's position can be
	// used to resume exactly from where it left off. Sources without
	// replay (TCP firehose, webhooks) must declare false so the
	// coordinator does not attempt gapless recovery on them.
	RecoverablePosition bool
}

Capabilities declares what a source supports: snapshot via DBLog, chunk SELECT introspection, and/or replication streaming. It is registration data — static per kind, not per instance — so the driver registry holds it and the orchestrator reads it without opening a source.

type Chunk ¶

type Chunk struct {
	Low  []any
	High []any
}

Chunk is a half-open primary-key range [Low, High). Low/High are tuples in PK column order.

type ChunkSource ¶

type ChunkSource interface {
	// Bounds returns the ordered chunk boundary keys.
	Bounds(ctx context.Context) ([][]any, error)
	// Scan reads one chunk, calling fn per row keyed by column name.
	Scan(ctx context.Context, ch Chunk, fn func(row map[string]any) error) error
}

ChunkSource is the chunk SELECT surface the snapshot orchestrator consumes. The SQL-backed chunkers implement it.

type DestructiveDDL ¶ added in v0.5.0

type DestructiveDDL struct {
	// Source is the driver kind, e.g. "postgres" or "mysql".
	Source string
	// Kind classifies the statement: "truncate" or "ddl".
	Kind string
	// Table is the affected table ("schema.table" when known; "" otherwise).
	Table string
	// Detail is the statement text or the affected-relation list.
	Detail string
}

DestructiveDDL is a TRUNCATE or other destructive DDL statement the source stream carried that the engine does not propagate to the sink. Reporting it lets the engine surface the divergence (a metric and a run event) or fail the run, instead of leaving the sink silently out of sync.

type Discoverer ¶ added in v0.3.0

type Discoverer interface {
	// Discover returns every table the connected user may SELECT, limited to
	// schemas when non-empty.
	Discover(ctx context.Context, schemas []string) ([]TableRef, error)
}

Discoverer is the optional source capability that lists the tables the pipeline should replicate when the spec omits an explicit table list (spec.Source.Postgres.Discover). The returned refs carry empty primary keys — Introspect derives them — and only Source is populated; the orchestrator derives Target from the sink namespace.

type Drainer ¶ added in v0.5.0

type Drainer interface {
	Drain(ctx context.Context, emit func(*dataplane.Batch) error) error
}

Drainer is an optional Reader capability: it flushes every event the decoder has already produced — buffered or still on its channel — into emit, without blocking for new data. The collapsed runner's relay uses it to flush decoded events ahead of a Closes marker, so an event decoded but not yet pulled can never be overtaken by the marker (issue #488).

type IncrementalSource ¶ added in v0.3.0

type IncrementalSource interface {
	Incremental(ctx context.Context, t TableRef, cursor string, after string) (next string, rows []map[string]any, more bool, err error)
}

IncrementalSource is the optional source capability for tables in incremental mode (spec.Table.Mode == "incremental"): a cursor-bounded read with no replication slot. Incremental returns at most one PAGE of rows (so a large table never loads entirely into memory), the cursor to resume from after that page, and whether more rows are already available (issue #572). The caller loops pages until more is false, then repeats on its interval. The cursor travels as a string; the source coerces it against the cursor column's type. The rows are the source's public row universe; the engine encodes them at the boundary. A source that can encode the page itself also implements BatchIncrementalSource (#733), which the engine prefers.

type Introspector ¶

type Introspector interface {
	Introspect(ctx context.Context, t spec.Table) (core.TableRef, core.Schema, []core.Warning, error)
}

Introspector resolves one spec table into its ref and the SOURCE canonical schema — the shape as the source emits it, before any cast or metadata column is applied. The pipeline resolves the cast policy centrally (the sink is the cast executor), so the source never applies it. The warnings are source-level advisories; the cast warnings come from the resolver.

type Mode ¶

type Mode uint8

Mode declares the synchronization mode a source supports.

const (
	// ModeCDC is log-based change data capture — the only mode implemented.
	ModeCDC Mode = iota
	// ModeIncremental is cursor-column incremental (e.g. updated_at): a
	// cursor-bounded read with no replication slot, serving sources without
	// replication access (RDS without grants, replicas without binlog). It
	// misses deletes and needs a cursor column. Implemented by Postgres.
	ModeIncremental
	// ModeBackfillOnly is a single snapshot with no stream. Declared in the
	// contract; not implemented.
	ModeBackfillOnly
)

type PartitionSource ¶

type PartitionSource interface {
	// Partitions returns n contiguous ranges covering the table's whole
	// primary-key domain, in order (Partitions(ctx,1)[0] is always the
	// single unbounded range). Returns an error if the primary key shape
	// isn't supported for range partitioning (e.g. more than one column).
	Partitions(ctx context.Context, n int) ([]Chunk, error)
}

PartitionSource is the optional chunker capability that samples a table's primary-key domain into n contiguous, ordered ranges.

The coordinator no longer requires or calls it: partition ownership is the rendezvous hash (internal/routing), decided from the key, and the snapshot reads by the chunker's own index ranges (ChunkSource.Bounds/Scan). It stays as the source-side capability for callers that need contiguous key ranges. Failing to implement it no longer blocks partitioning.

type Positioner ¶

type Positioner interface {
	// InitialPosition is the stream start for a first boot (no resume).
	InitialPosition(ctx context.Context) (position.Position, error)
	// ParsePosition decodes a stored cdc.position string.
	ParsePosition(s string) (position.Position, error)
}

Positioner resolves the stream start position: the first-boot start and the codec for a stored cdc.position.

type QuerySource ¶

type QuerySource interface {
	// NewChunker builds the chunk SELECT source for one table.
	NewChunker(source, pk string, chunkSize int) (ChunkSource, error)
	// CloseQuery releases the source's query connection.
	CloseQuery() error
}

QuerySource is the SQL-backed surface of a relational source: the chunk SELECT source for one table and the query-connection lifecycle. MySQL and Postgres implement it; a stream source (kafka) does not. The source owns its query connection — opened at driver construction, released on CloseQuery — so the orchestration never touches *sql.DB.

type Reader ¶

type Reader interface {
	SourceReader
	// Start begins replication from `from`. Called once before the first
	// Next. Sources that need a resume position (MySQL GTID set, Postgres
	// LSN, Kafka offsets) apply it here; sources without replay ignore it.
	Start(ctx context.Context, from position.Position) error
	// Next returns the next columnar batch, or (nil, nil) at a clean end.
	// The caller owns the returned batch and must Release() it.
	Next(ctx context.Context) (*dataplane.Batch, error)
	Close()
	// SetConfirmed installs a callback that returns the minimum position
	// committed to the sink across all tables. The Postgres reader uses
	// this to advance the slot's confirmed_flush_lsn only to the point
	// that has been durably written — never past it.
	SetConfirmed(f func() position.Position)
}

Reader is the replication reader surface a driver drives: the DBLog watermark surface plus the columnar pull stream. A Reader is a handle obtained from Source.Open; the stream is pulled via Next (M4: sources produce Arrow RecordBatches, one batch per call).

type Runtime ¶

type Runtime struct {
	ServerID  uint32
	Heartbeat time.Duration
	Logger    *slog.Logger
	// OnDestructiveDDL, when non-nil, is called for every TRUNCATE or other
	// destructive DDL the reader sees on the stream. The engine owns what
	// happens next — a metric and a run event under the default policy, or a
	// failed run under onTruncate: fail. The reader does not decide. Nil is a
	// no-op.
	OnDestructiveDDL func(DestructiveDDL)
}

Runtime carries the replication knobs a driver passes through to the source reader.

type SchemaSetter ¶

type SchemaSetter interface {
	SetSourceSchemas(schemas map[string]core.Schema)
}

SchemaSetter is an optional Reader capability: the coordinator/runner holds the RESOLVED canonical schema (spec-declared for stream sources, cast-applied) and injects it after Open. The source boundary then gates on drift against the native row shape and encodes stable batches — never per-drain inference.

type Source ¶

type Source interface {
	Streamer
	Positioner
	Introspector
}

Source is the minimal surface every source provides: it streams changes, resolves positions, and introspects schemas. It deliberately carries NO SQL shape; the SQL-only surface (the chunked snapshot SELECT) lives on QuerySource, which only relational sources implement, so a stream source never stubs methods it cannot do.

type SourceReader ¶

type SourceReader interface {
	Synced() position.Position
	Master(ctx context.Context) (position.Position, error)
	// OpenWindow starts the DBLog window for chunkID: the reader captures
	// its own source watermark and, from now on, tags decoded events whose
	// position is strictly past that watermark InWindow for the chunk.
	// Events at or before the watermark are already reflected in the chunk
	// SELECT and must not be tagged. ClearWindow ends the window.
	OpenWindow(ctx context.Context, chunkID uint32)
	ClearWindow()
}

SourceReader is the reader surface the snapshot orchestrator needs: positions for the watermarks, and the window that tags events InWindow at decode time — so no event can escape the window by racing a channel pull.

type Streamer ¶

type Streamer interface {
	Open(ctx context.Context, refs []TableRef) (Reader, error)
}

Streamer opens the replication reader over the pipeline's tables. The source owns its own connections; the reader carries no SQL shape.

type TableRef ¶

type TableRef = core.TableRef

TableRef maps one source table to its target and primary key. It is an alias of the canonical core.TableRef — the pipeline-wide table identity.