core

type CastPolicy ¶

type CastPolicy struct {
	Columns map[string]CastTarget
}

CastPolicy is the per-table map of source column name → target canonical type. Keys are source-side names, matching the introspection result.

func ParseCastPolicy(m map[string]string) (CastPolicy, error)

func (p CastPolicy) Resolve(src Schema) (Schema, []Warning, error)

func (p CastPolicy) Target(name string) (CastTarget, bool)

type CastTarget ¶

type CastTarget struct {
	Type ColumnType
	// AssumeUTC marks timestamp → timestamptz(assume_utc): the naive
	// literal is asserted to be UTC, never shifted. Required for that cast.
	AssumeUTC bool
	// Encoding names the explicit encoding for binary → string: "hex" or
	// "base64". Plain binary → string is ambiguous and prohibited.
	Encoding string
}

CastTarget is the resolved destination of a per-column cast: a canonical type plus the explicit rules the matrix requires for the ambiguous cases.

func ParseCastTarget(s string) (CastTarget, error)

func (t CastTarget) Convert(from Kind, v any) (any, error)

func (t CastTarget) String() string

type Column ¶

type Column struct {
	Name string
	Type ColumnType
}

Column is one named canonical column.

type ColumnType ¶

type ColumnType struct {
	Kind      Kind
	Precision int
	Scale     int
	FixedSize int
	Nullable  bool
	// Opaque records the provenance of an unmappable source type: what it
	// was and where it came from, so the validation error and any operator
	// looking at it know what the escape valve is carrying. Non-nil only
	// when Kind == KindUnknown.
	Opaque *OpaqueOrigin

	// Fields holds the ordered fields of a KindStruct column.
	Fields []Column
	// Elem is the element type of a KindList column.
	Elem *ColumnType
	// KeyType and ValueType are the key/value types of a KindMap column.
	// Message-log map keys are almost always strings.
	KeyType   *ColumnType
	ValueType *ColumnType
}

ColumnType is a canonical column type. Precision/Scale are only meaningful for KindDecimal; FixedSize only for KindFixedBinary; Opaque only when Kind is KindUnknown; the composite fields only when Kind is KindStruct (Fields), KindList (Elem) or KindMap (KeyType/ValueType). The composite fields are pointers so a Struct can nest a Struct, or a List hold a Struct, without a parallel type system.

func ParseColumnType(s string) (ColumnType, error)

func (t ColumnType) String() string

type Kind ¶

type Kind uint8

Kind is the canonical type.

The admission criterion is deliberate: a value enters the Kind set only if a supported source produces it natively AND a supported sink has a real representation for it. Physical encodings of the same logical type (dictionary, *_view, large_*) never become distinct Kinds; polymorphic types with no source or destination (union, extension) never enter; and the escape valve (KindUnknown + an explicit cast) is the deliberate alternative to accepting anything silently. Composite types (struct/list/map) are a separate dimension of the model — they touch the cast matrix, the Arrow mapping, Iceberg schema generation, the wire format and schema comparison — and land as a dedicated effort, not an enum addition.

const (
	KindUnknown Kind = 0
	KindBool    Kind = 1
	KindInt32   Kind = 2
	KindInt64   Kind = 3
	// KindUInt64 is an unsigned 64-bit integer; sources that natively emit
	// uint64 keep their exact width instead of silently widening to int64.
	KindUInt64  Kind = 4
	KindFloat32 Kind = 5
	KindFloat64 Kind = 6
	// KindDecimal uses Precision and Scale.
	KindDecimal Kind = 7
	KindString  Kind = 8
	KindBinary  Kind = 9
	// KindFixedBinary is Iceberg fixed(L) — a fixed-size byte sequence,
	// distinct from variable binary.
	KindFixedBinary Kind = 10
	// KindDate is days since epoch.
	KindDate Kind = 11
	// KindTime is micros since midnight.
	KindTime Kind = 12
	// KindTimestamp is a naive wall clock, NO timezone (MySQL DATETIME).
	KindTimestamp Kind = 13
	// KindTimestampTZ is a UTC instant (MySQL TIMESTAMP).
	KindTimestampTZ Kind = 14
	KindUUID        Kind = 15
	// KindJSON is semantically JSON; physically a string in most sinks.
	KindJSON Kind = 16

	// Composite kinds — a separate dimension of the model, entered only
	// because both sides of the boundary have a real representation for
	// them: message-log sources (Avro/Protobuf via schema registry) produce
	// nested values and Iceberg holds struct/list/map natively.
	KindStruct Kind = 17
	KindList   Kind = 18
	KindMap    Kind = 19
)

func (k Kind) String() string

type MetadataColumn ¶

type MetadataColumn struct {
	From MetadataKey `json:"from"`
	As   string      `json:"as"`
}

MetadataColumn maps one metadata key to a destination column name. The destination name is explicit — there is no default, so an internal wire name never leaks into the user's schema.

type MetadataKey ¶

type MetadataKey string

MetadataKey identifies one pipeline-provided value that can be landed as a column in the target table. The catalog is closed: every key has a fixed canonical type and a fixed semantic, and there is no free-form field.

const (
	// MetaOp is the row operation: "insert", "update" or "delete". In upsert
	// mode a delete removes the row entirely, so op never lands as "delete"
	// there — the row stops existing. In append mode every event becomes a
	// row and delete survives in op.
	MetaOp MetadataKey = "op"
	// MetaCommitTS is when the source committed the transaction that
	// produced the row. Snapshot rows carry NULL: the chunk SELECT does not
	// return the creating transaction's commit time.
	MetaCommitTS MetadataKey = "commit_ts"
	// MetaIngestTS is when the pipeline processed the event. The difference
	// between commit_ts and ingest_ts is the replication lag, measured per
	// row at the destination.
	MetaIngestTS MetadataKey = "ingest_ts"
	// MetaPosition is the row's source coordinate (GTID set, LSN, or Kafka
	// offset map). Snapshot rows carry the chunk's low watermark.
	MetaPosition MetadataKey = "position"
	// MetaSourceTable is the source table the row came from, e.g.
	// "shop.orders".
	MetaSourceTable MetadataKey = "source_table"
	// MetaPhase distinguishes backfill from live traffic: PhaseSnapshot for
	// rows read by the DBLog chunk SELECT, PhaseStream for live events. It is
	// an axis orthogonal to op — a snapshot row is semantically an insert.
	MetaPhase MetadataKey = "phase"

	// Transport metadata — the message-queue envelope (Kafka today, Kinesis
	// next). Names are transport-neutral so they do not need renaming as
	// sources are added. For CDC sources only MetaStream and MetaSequence
	// carry values (the source table and the GTID/LSN of the event); the
	// rest are NULL.
	MetaStream  MetadataKey = "stream"   // topic / stream name (CDC: source table)
	MetaShard   MetadataKey = "shard"    // partition / shard id (CDC: NULL)
	MetaSeq     MetadataKey = "sequence" // offset / sequence number (CDC: GTID/LSN of the event)
	MetaMsgTS   MetadataKey = "msg_ts"   // message arrival time (CDC: NULL)
	MetaMsgKey  MetadataKey = "msg_key"  // message key (CDC: NULL)
	MetaHeaders MetadataKey = "headers"  // JSON-serialized headers (CDC: NULL)
)

func (k MetadataKey) ColumnType() ColumnType

func (k MetadataKey) String() string

func (k *MetadataKey) UnmarshalJSON(b []byte) error

func (k *MetadataKey) UnmarshalText(b []byte) error

type OpaqueOrigin ¶

type OpaqueOrigin struct {
	TypeName   string
	VendorName string
}

OpaqueOrigin names an unmappable source type and its vendor. It is the provenance half of the escape valve: KindUnknown says "this has no canonical form and needs an explicit cast", Opaque says what it was ("geometry" from "postgres", "point" from "mysql") instead of discarding that knowledge at the mapping boundary.

func (o *OpaqueOrigin) String() string

type Row ¶

type Row map[string]any

Row is a decoded source row: column name → value. The Go type per Kind is the canonical contract both cast kernels (core.Convert and the columnar encoder) rely on:

KindBool        bool
KindInt32       int32
KindInt64       int64
KindUInt64      uint64
KindFloat32     float32
KindFloat64     float64
KindDecimal     string (decimal text)
KindString      string
KindBinary      []byte
KindFixedBinary []byte
KindDate        time.Time, int32 days, or "2006-01-02" text
KindTime        time.Time, int64 micros, or "15:04:05" text
KindTimestamp   time.Time or naive timestamp text
KindTimestampTZ time.Time or RFC3339 text
KindUUID        string or 16 raw bytes
KindJSON        string, []byte, or a Go composite (map/slice)

A Kind may arrive in more than one representation — the DBLog snapshot and the live stream decode differently — so both kernels accept the union.

type Schema ¶

type Schema struct {
	Columns    []Column
	PrimaryKey []string // column names, in key order
}

Schema is the canonical table shape — the ONLY schema type that crosses the source↔sink boundary.

func FilterSchemaColumns(cs Schema, columns []string) Schema

func WireSchema(source, resolved Schema) Schema

func (s Schema) Column(name string) (Column, bool)

func (s Schema) KeyIndexes() ([]int, error)

func (s Schema) Validate() error

type TableRef ¶

type TableRef struct {
	Source     string   // source-side identifier, e.g. "shop.orders"
	Target     string   // sink-side identifier, e.g. "raw.orders"
	PrimaryKey []string // equality key; empty means "derive from source"
	// Owner is the worker group that owns this table's partition
	// ({pipeline}-{target}-{i}, spec.Table.WorkerGroupNames). Empty when the
	// table is not partitioned. Sinks that persist a durable position per
	// table use it to keep one position per partition instead of a single
	// last-writer scalar (WK-001 §2.6). Derived from the spec, so it is
	// stable across restarts and k8s rollouts — unlike the pod name.
	Owner string
	// OwnerCount is the number of partitions the table is split across
	// (spec.Table.WorkerCount). A per-partition Position() read uses it to
	// refuse a MinSafe over an incomplete owner set: an owner with no
	// committed position would otherwise be resumed past. 0 or 1 means the
	// collapsed single-owner path.
	OwnerCount int
}

TableRef identifies one replicated table on both sides. This is the pipeline-wide table identity (replaces the per-source TableRef).

type Warning ¶

type Warning struct {
	Message string
}

Warning is a validation outcome that does not reject the spec but must be surfaced to the operator (eventlog, status) — never only in logs.