Help us shape the future of Go! Take the 2026 Go Developer Survey

rowmodel

package
v0.31.0 Latest Latest
Warning

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

 Go to latest Published: Oct 28, 2026 License: Apache-2.0 Imports: 9 Imported by: 0

Documentation

Overview

Package rowmodel defines the connector-neutral row schema and per-row metadata. It deliberately has no dependency on Arrow or the root filament package.

Index

Constants

View Source 
const (
	AuditRunIDField          = "_filament_run_id"
	AuditRunStartedAtField   = "_filament_run_started_at"
	AuditOperationField      = "_filament_operation"
	AuditSourcePositionField = "_filament_source_position"
	AuditSequenceField       = "_filament_sequence"
)

Filament reserves this namespace for row lineage materialized after source fields. Audit fields remain nullable so existing destination tables can evolve without rewriting historical rows.

View Source 
const (
	EventIDField        = "_filament_event_id"
	EventIdentityField  = "_filament_event_identity"
	EventTimestampField = "_filament_event_ts"
	EventHeadersField   = "_filament_headers"
	EventKeyField       = "_filament_key"
	EventPayloadField   = "_filament_payload"
)

Reserved envelope column names identify row-aligned event metadata.

Variables

View Source 
var ErrPositionIncomparable = errors.New("progress: positions incomparable")

ErrPositionIncomparable indicates that compatible progress ordering cannot be established.

Functions

func IsReservedColumn added in v0.21.0 

func IsReservedColumn(name string) bool

IsReservedColumn reports whether name sits in the prefix Filament keeps for its own audit columns.

func ValidateReservedFields added in v0.6.0 

func ValidateReservedFields(s Schema) error

ValidateReservedFields permits only the exact SDK-declared suffix. The marker is schema provenance, not a security boundary against malicious Go code.

Types

type Checkpoint 

type Checkpoint interface {
	Resource() string
	Int(key string) int
	String(key string) string
	Set(key string, value any) Checkpoint
	Raw() map[string]any
}

Checkpoint is a resource's immutable resumable cursor.

type CheckpointData 

type CheckpointData struct {
	ResourceName string         `json:"resource"`
	Cursor       map[string]any `json:"cursor"`
}

CheckpointData is the concrete checkpoint representation.

func NewCheckpoint 

func NewCheckpoint(resource string) *CheckpointData

NewCheckpoint returns an empty checkpoint for resource.

func (*CheckpointData) Int 

func (c *CheckpointData) Int(key string) int

Int reads key as an integer, accepting common JSON numeric representations.

func (*CheckpointData) Raw 

func (c *CheckpointData) Raw() map[string]any

Raw returns the cursor payload used by persistence adapters.

func (*CheckpointData) Resource 

func (c *CheckpointData) Resource() string

Resource returns the resource this checkpoint advances.

func (*CheckpointData) Set 

func (c *CheckpointData) Set(key string, value any) Checkpoint

Set returns an independent checkpoint with key updated.

func (*CheckpointData) String 

func (c *CheckpointData) String(key string) string

type CodecResolver added in v0.6.0 

type CodecResolver interface {
	Lookup(codec string, version int) (PositionCodec, error)
}

CodecResolver locates a position codec by identifier and version.

type Control added in v0.6.0 

type Control struct {
	Kind               ControlKind
	Domain             DomainKey
	Position           Position
	TxnID              string
	Resource           string
	MembershipRevision int64
}

Control is a standalone source boundary, including boundaries with no rows. It supplies source evidence, not proof of pipeline completion or durability.

func (Control) Clone added in v0.6.0 

func (c Control) Clone() Control

Clone returns a control with independently owned position bytes.

func (Control) Validate added in v0.6.0 

func (c Control) Validate() error

Validate checks the control kind and its required source-boundary fields.

type ControlKind added in v0.6.0 

type ControlKind uint8

ControlKind identifies a standalone source boundary marker. 

const (
	ControlUnspecified ControlKind = iota
	TxnBegin
	TxnEnd
	ProgressBoundary
	MemberActivated
)

Control kinds distinguish source transactions, progress, and membership boundaries.

type DomainKey added in v0.6.0 

type DomainKey struct{ Incarnation, Domain string }

DomainKey names a source ordering domain within a stable source incarnation. Consumer, run and attempt identities must not be used as incarnations.

func (DomainKey) Validate added in v0.6.0 

func (d DomainKey) Validate() error

Validate requires nonempty UTF-8 incarnation and domain identifiers.

type DomainPosition added in v0.6.0 

type DomainPosition struct {
	Domain   DomainKey
	Position Position
}

DomainPosition pairs a source domain with its bookmark for wire serialization.

type DomainPositions added in v0.6.0 

type DomainPositions map[DomainKey]Position

DomainPositions uses sorted entries on the wire; duplicate domains are invalid. Clone when handing ownership across a session/coordinator boundary.

func (DomainPositions) Clone added in v0.6.0 

func (p DomainPositions) Clone() DomainPositions

Clone returns an independently owned map and position values.

func (DomainPositions) Entries added in v0.6.0 

func (p DomainPositions) Entries() ([]DomainPosition, error)

Entries validates and copies positions into deterministic incarnation/domain order.

func (DomainPositions) MarshalJSON added in v0.6.0 

func (p DomainPositions) MarshalJSON() ([]byte, error)

MarshalJSON encodes sorted entries instead of a map with struct keys.

func (*DomainPositions) UnmarshalJSON added in v0.6.0 

func (p *DomainPositions) UnmarshalJSON(data []byte) error

UnmarshalJSON validates entries and rejects duplicate domains before replacing the map.

type EventIdentity added in v0.6.0 

type EventIdentity struct {
	Domain   DomainKey
	Position Position
	Ordinal  uint64
}

EventIdentity identifies one source event independently of its run, attempt, generation or consumer. Ordinal distinguishes events sharing a position.

func (EventIdentity) Clone added in v0.6.0 

func (e EventIdentity) Clone() EventIdentity

Clone returns an event identity with independently owned position bytes.

type Field 

type Field struct {
	Name      string
	Nullable  bool
	Logical   LogicalType
	Native    string
	Precision int
	Scale     int
}

Field describes one portable column and its source-native spelling.

type LogicalType 

type LogicalType string

LogicalType is a portable, engine-independent column type. 

const (
	LogicalUnknown     LogicalType = ""
	LogicalBool        LogicalType = "bool"
	LogicalInt16       LogicalType = "int16"
	LogicalInt32       LogicalType = "int32"
	LogicalInt64       LogicalType = "int64"
	LogicalFloat32     LogicalType = "float32"
	LogicalFloat64     LogicalType = "float64"
	LogicalDecimal     LogicalType = "decimal"
	LogicalString      LogicalType = "string"
	LogicalBytes       LogicalType = "bytes"
	LogicalDate        LogicalType = "date"
	LogicalTime        LogicalType = "time"
	LogicalTimestamp   LogicalType = "timestamp"
	LogicalTimestampTZ LogicalType = "timestamptz"
	LogicalJSON        LogicalType = "json"
	LogicalUUID        LogicalType = "uuid"
	LogicalArray       LogicalType = "array"
)

Portable column types.

type Meta 

type Meta struct {
	Op     Operation
	Key    []string
	Coarse bool
	LSN    string
	Seq    uint64
	Stream *StreamMeta
}

Meta accompanies a row without becoming an Arrow column.

type Operation 

type Operation uint8

Operation is the change kind carried by a row. 

const (
	OpInsert Operation = iota
	OpUpdate
	OpDelete
)

Row operations.

type Position added in v0.6.0 

type Position struct {
	Codec   string
	Version int
	// Value is the codec-encoded source position, not message content.
	Value []byte
}

Position stores a codec-encoded source bookmark, independently of message content.

func CanonicalPosition added in v0.6.0 

func CanonicalPosition(r CodecResolver, p Position) (Position, error)

CanonicalPosition validates and canonicalizes a position without retaining caller buffers.

func (Position) Clone added in v0.6.0 

func (p Position) Clone() Position

Clone returns a position with independently owned encoded bytes.

func (Position) Validate added in v0.6.0 

func (p Position) Validate() error

Validate checks codec identity and version; the codec validates Value.

type PositionCodec added in v0.6.0 

type PositionCodec interface {
	Validate(Position) error
	Compare(a, b Position) (PositionOrder, error)
	Canonicalize(Position) (Position, error)
}

PositionCodec validates, compares, and canonicalizes one source position format. Implementations must be deterministic, pure, and safe for concurrent use.

type PositionOrder added in v0.6.0 

type PositionOrder uint8

PositionOrder describes codec-defined progress ordering; zero fails closed. Encoded byte ordering does not establish source progress. 

const (
	PositionIncomparable PositionOrder = iota
	PositionEqual
	PositionBefore
	PositionAfter
)

Position comparison outcomes fail closed when ordering cannot be established.

func ComparePositions added in v0.6.0 

func ComparePositions(r CodecResolver, aDomain DomainKey, a Position, bDomain DomainKey, b Position) (PositionOrder, error)

ComparePositions compares compatible domains and codecs, failing closed on incomparable progress.

type ResourceDomainRef added in v0.6.0 

type ResourceDomainRef struct {
	Resource string
	Domain   DomainKey
}

ResourceDomainRef describes a relationship, not an independently writable per-resource checkpoint. Multiple resources may share the same log domain.

type Schema 

type Schema struct {
	Resource string
	// DestinationResource names the output object without changing source identity.
	// Empty preserves the source resource name.
	DestinationResource string
	Fields              []Field
	PrimaryKey          []string
	Engine              string
	// contains filtered or unexported fields
}

Schema is one resource's column layout.

func AsCDCAppendHistory 

func AsCDCAppendHistory(schema Schema) Schema

AsCDCAppendHistory returns the sink schema used to retain every CDC event. Source keys may repeat and deletes may contain only key values, so the destination has no primary key and every source field is nullable.

func WithAuditFields 

func WithAuditFields(schema Schema, cdc bool) (Schema, error)

WithAuditFields returns an independently owned destination schema with Filament's universal lineage fields and, for CDC, change-stream fields.

func WithEnvelopeFields added in v0.6.0 

func WithEnvelopeFields(s Schema) (Schema, error)

WithEnvelopeFields declares the SDK-owned suffix. It rejects user collisions before adding columns; callers cannot mark arbitrary fields as SDK-generated.

func WithEventMetadataFields added in v0.6.0 

func WithEventMetadataFields(s Schema) (Schema, error)

WithEventMetadataFields adds event metadata without the raw message payload. Decoded message columns remain source-owned.

func (Schema) Clone 

func (s Schema) Clone() Schema

Clone returns an independently owned schema. Connectors commonly reuse and mutate discovery buffers, so registries must not retain their backing slices.

func (Schema) Equal 

func (s Schema) Equal(other Schema) bool

Equal reports whether two supplied schemas describe the same resource model.

type StreamMeta added in v0.6.0 

type StreamMeta struct{ Identity EventIdentity }

StreamMeta carries native event metadata. Its presence does not select continuous execution: bounded catch-up can also carry event identities. It cannot certify coverage or replace the row-aligned envelope columns.

func (*StreamMeta) Clone added in v0.6.0 

func (m *StreamMeta) Clone() *StreamMeta

Clone copies stream metadata and its position bytes, preserving nil metadata.

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.