Record model

The record model is the currency every connector and operator speaks. A connector maps its native types onto a small, storage-agnostic type set; from that point on, the engine moves typed data. Because the schema is explicit and column order is fixed, a transform or a sink never has to inspect a value to learn its type at runtime — the type is already known, and a value that doesn't match fails fast instead of being silently coerced. The package comment says it plainly: this eliminates the class of silent type-coercion bugs common in dynamically-typed (pandas/polars-style) flows.

The type set

DataType is a logical type, independent of any storage engine. There are exactly seven, and connectors map their native types onto these:

// DataType is the logical type of a field. It is storage-engine agnostic;
// connectors map their native types onto these.
type DataType string

const (
	TypeString    DataType = "string"
	TypeInt64     DataType = "int64"
	TypeFloat64   DataType = "float64"
	TypeBool      DataType = "bool"
	TypeTimestamp DataType = "timestamp"
	TypeBytes     DataType = "bytes"
	TypeJSON      DataType = "json"
)

Seven is a deliberate floor, not a limitation. A wide-open type universe pushes complexity onto every connector, every operator, and every sink — each has to know how to handle each type. A small closed set means the hard type mapping happens once, at the connector edge (Postgres numeric → float64, Mongo ObjectId → string, a document column → json), and every stage after that reasons over the same seven cases. json is the escape hatch for structure the type set doesn't name: it rides as a string so it moves losslessly, and operators like jsonextract/flatten project fields out of it when needed.

Schema and Field

A Field is a name, a type, and a nullable flag. A Schema is an ordered set of fields with unique, non-empty names. Order matters: it is the contract that lets rows store values positionally. NewSchema validates uniqueness and non-empty names and types, returning an error rather than panicking, so schema discovery from a live source can surface the problem cleanly. An internal name → position index makes IndexOf lookups O(1).

// Field describes one column.
type Field struct {
	Name     string   `json:"name"`
	Type     DataType `json:"type"`
	Nullable bool     `json:"nullable"`
}

// Schema is an ordered set of fields with unique names.
type Schema struct {
	Fields []Field `json:"fields"`
	index  map[string]int // name → position, for O(1) IndexOf (unexported)
}

// NewSchema validates that names are unique and non-empty and types are set,
// returning an error (not a panic) so live schema discovery can surface it.
func NewSchema(fields ...Field) (*Schema, error)
func (s *Schema) IndexOf(name string) int // column position, or -1
func (s *Schema) Len() int

There is a subtle, important detail in that index field: it is unexported, so encoding/json never touches it. A schema that arrives over the wire — deserialized from a pipeline document, a blob manifest, or an API payload — has Fields populated but a nil index. IndexOf handles this by lazily rebuilding the index on first use, so name lookups work identically whether a schema was constructed with NewSchema or decoded from JSON. The positional Fields slice is the durable truth; the map is a rebuildable accelerator.

func (s *Schema) IndexOf(name string) int {
	if s.index == nil { // e.g. schema was JSON-decoded, index not populated
		s.rebuildIndex()
	}
	if i, ok := s.index[name]; ok {
		return i
	}
	return -1
}

Row: positional values

A Row is a pointer to its Schema plus a slice of Values aligned positionally to that schema's fields. There are no field names stored per row — value i is field i. A nil entry represents SQL NULL. NewRow checks that the value count matches the schema arity, so a malformed row can't enter the pipeline; Get(name) is a convenience that resolves a name through the schema index.

// Row is a single record: values aligned positionally to a Schema's fields.
// A nil entry represents SQL NULL.
type Row struct {
	Schema *Schema
	Values []any
}

// NewRow checks the value count matches the schema arity.
func NewRow(schema *Schema, values ...any) (Row, error)

// Get returns the value for the named column and whether it exists.
func (r Row) Get(name string) (any, bool)

Why positional and not a map[string]any? Three reasons, all structural. A map stores every column name on every row — for a million rows that is a million redundant copies of the same keys, plus a hash-table header per row. A slice stores just the values, once, with the names living a single time on the shared schema. Iteration over a slice is a cache-friendly linear walk; map iteration chases pointers in arbitrary order. And a map silently tolerates a missing or extra key, whereas NewRow's arity check makes a shape mismatch a loud error at the boundary. The one cost is that values are any, so scalars are boxed — but the type is recoverable in O(1) from the schema, which is exactly what ValidateValue relies on below.

Batch: the unit of movement

Rows never move one at a time. A Batch is a chunk of rows that all share one schema, and the batch is what a RecordIterator yields, what an operator's Process transforms, and what a Sink.Write persists. Sharing a single schema pointer across the whole batch keeps rows compact and makes the common operation — iterate every row, touch column i — a tight positional loop.

// Batch is a chunk of rows sharing one schema. Engines move data in batches
// to amortise per-row overhead and to spill cleanly to object storage.
type Batch struct {
	Schema *Schema
	Rows   []Row
}

func (b *Batch) Len() int { return len(b.Rows) }

Two properties fall out of positional, single-schema batches, and both matter at scale:

• Amortized per-row cost. The schema is resolved once for the batch, not once per row. Type dispatch, column offsets, and target preparation happen at batch granularity, so the marginal cost of one more row is just its values. Batch size is the knob that trades memory for per-call overhead — a source picks a batch size (Postgres reads a configurable chunk per Next), and the runner's memory ceiling is roughly one batch in flight.
• Clean spill. Because a batch is "these rows, this one schema," it serializes to object storage as a self-contained chunk: write the schema once in a manifest, then stream the rows as compact lines. That is exactly what the blob staging layer does — the record model was shaped to make that spill trivial.

How a batch spills: the JSONL codec

The staging connector serializes a batch to gzipped JSONL, one JSON object per row, and it leans directly on the type set to round-trip Go types through JSON without loss. Timestamps become RFC3339Nano strings (UTC), bytes become base64 strings, and everything else is JSON-native. On the way back, coerce undoes it — critically, encoding/json decodes all numbers as float64, so an int64 column is explicitly re-narrowed rather than left as a float.

func coerce(t record.DataType, raw any) (any, error) {
	switch t {
	case record.TypeInt64:
		f, ok := raw.(float64) // encoding/json numbers decode as float64
		if !ok { return nil, fmt.Errorf("want number, got %T", raw) }
		return int64(f), nil
	case record.TypeTimestamp:
		s, _ := raw.(string)
		ts, err := time.Parse(time.RFC3339Nano, s) // canonical text form
		if err != nil { return nil, err }
		return ts.UTC(), nil
	case record.TypeBytes:
		s, _ := raw.(string)
		return base64.StdEncoding.DecodeString(s)  // base64 → []byte
	// string/json → string, float64 → float64, bool → bool …
	}
}

This is the payoff of a closed type set: the codec is a single switch over seven cases. A row that spilled at the edge and a row that came straight from Postgres are the same in-memory shape by the time an operator sees them, so nothing downstream can tell whether a run was direct or staged.

Failing fast: ValidateValue

The payoff of an explicit type system is the ability to reject bad data before it lands. ValidateValue checks that a value is assignable to its declared type, allowing nil only when the field is nullable. Sinks and transforms call it (via connector.ValidateBatch) to fail loudly instead of writing a silently coerced or garbage value.

// ValidateValue checks that v is assignable to t (nil allowed only when
// nullable). Sinks/transforms use it to fail fast on bad data.
func ValidateValue(t DataType, nullable bool, v any) error {
	if v == nil {
		if nullable {
			return nil
		}
		return fmt.Errorf("nil value for non-nullable %s", t)
	}
	ok := false
	switch t {
	case TypeString, TypeJSON:
		_, ok = v.(string)
	case TypeInt64:
		_, ok = v.(int64)
	case TypeFloat64:
		_, ok = v.(float64)
	case TypeBool:
		_, ok = v.(bool)
	case TypeTimestamp:
		_, ok = v.(time.Time)
	case TypeBytes:
		_, ok = v.([]byte)
	default:
		return fmt.Errorf("unknown data type %q", t)
	}
	if !ok {
		return fmt.Errorf("value %v (%T) is not a valid %s", v, v, t)
	}
	return nil
}

ValidateBatch layers over it: it confirms every row's value count matches the schema arity and validates each cell against its field, pointing at the exact row and column when something is wrong. Note that string and json share the same underlying Go type (string), so validation treats them alike — the json type is a semantic marker for operators, not a distinct runtime representation.

Cursors: the typed incremental watermark

Incremental replication is expressed entirely in terms of the record model. A Cursor is a typed watermark whose Kind mirrors a stream field's DataType — the engine tracks "the highest cursor value seen so far" and, on the next run, reads only rows at or beyond it. Only ordered types make sense as cursors: int64 (an autoincrement id), timestamp (an updated_at), or string.

// Cursor is a typed incremental watermark. Kind mirrors the stream field's type.
type Cursor struct {
	Kind  record.DataType
	Value any // int64 | time.Time | string
}

// ParseCursor turns a stored text watermark into a typed Cursor. Empty ⇒ nil.
func ParseCursor(kind record.DataType, text string) (*Cursor, error)
// FormatCursor renders a typed cursor to its canonical text form.
func FormatCursor(c *Cursor) string
// CompareCursor returns -1/0/1 for a<b / a==b / a>b under the given kind.
func CompareCursor(a, b any, kind record.DataType) int

The watermark is stored as text (it has to survive in a database column and travel in a JSON job spec) but compared as its typed value. That distinction is the whole point of FormatCursor/ParseCursor/CompareCursor: an int64 cursor round-trips through base-10 text and an RFC3339Nano timestamp round-trips through UTC text, so ordering is numeric or temporal, never lexical — "9" correctly precedes "10", and 2026-07 correctly precedes 2026-11, neither of which a string comparison would get right.

A source that can push the predicate down implements IncrementalSource (Postgres emits WHERE field >= $1 ORDER BY field); one that can't is wrapped in a clientFilterIterator that drops rows below the watermark and tracks the running max in memory. Both paths make the same deliberate choice about NULLs: a row whose cursor value is NULL or absent cannot be ordered, so it is excluded from the delta rather than silently counted as synced — the SQL field >= $1 already excludes NULLs, and the client filter fires an onNull callback so the host can warn.