core

type BufferedRowReporter ¶ added in v1.1.0

type BufferedRowReporter interface {
	BufferedRows() int
}

BufferedRowReporter is implemented by a sink that can say how many rows it is holding.

A sink keeps every row a failed flush could not deliver, and nothing bounds that buffer. Today a failed flush stops the pipeline, so the buffer dies with the process. That bound is a property of the call pattern rather than of the design, and it disappears the day a flush failure stops being fatal. An operator needs to see the depth before that day, not after.

Optional, like MarkCommitter: a sink that buffers nothing reports nothing.

type ErrorPolicy ¶

type ErrorPolicy int

const (
	PolicyRaise ErrorPolicy = iota
	PolicyIgnore
	PolicyDLQ
)

func ParseErrorPolicy(s string) (ErrorPolicy, error)

type Handler ¶

type Handler interface {
	Init(ctx context.Context) error
	Write(msg []byte) error
	Invoke(ctx context.Context) (arrow.Table, error)

	// RowsRead reports the rows the last Invoke ingested into the batch table.
	//
	// It is the denominator of the enrichment ratio, so it has to count rows
	// the SQL actually ran over. An Invoke that failed schema inference
	// produced no table at all and reports zero, however many messages were
	// written to it -- counting accepted writes instead would overstate the
	// denominator on exactly the path where rows were lost.
	RowsRead() int64
}

type Mark ¶ added in v1.0.4

type Mark struct {
	Offset      int64
	LeaderEpoch int32
}

Mark is the position of the last message the pipeline has finished with in one partition: written to the handler, or dropped by an error policy.

type MarkCommitter ¶ added in v1.0.4

type MarkCommitter interface {
	CommitMarks(marks *Marks) error
}

MarkCommitter is implemented by sources that can commit an explicit position. The pipeline prefers it to Commit, because a source that reads ahead of the pipeline -- the Kafka source polls into a buffer -- has fetched messages the pipeline has not processed, and committing "everything fetched" commits those too.

type Marks ¶ added in v1.0.5

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

Marks records the last position the pipeline has finished with in each partition it has read: a message written to the handler, or one dropped by an error policy.

Topic-then-partition nesting is the natural shape for this and the awkward one to pass around. Owning it here keeps two rules in one place rather than repeated at every call site: a partition only ever moves forward, and iteration is ordered so anything rendered from it -- the stats endpoint, the CLI -- is stable between runs.

func NewMarks() *Marks

func (m *Marks) Advance(topic string, partition int32, mark Mark)

func (m *Marks) Each(fn func(topic string, partition int32, mark Mark))

func (m *Marks) Empty() bool

func (m *Marks) Get(topic string, partition int32) (Mark, bool)

func (m *Marks) Len() int

type Message ¶

type Message struct {
	Value     []byte
	Topic     string
	Partition int32
	Offset    int64
	// LeaderEpoch is the Kafka leader epoch the record was read under. It is
	// carried through so a commit can name it, which lets the broker detect
	// log truncation. Only meaningful when HasMetadata is true; a source with
	// no positions leaves it zero along with the rest.
	LeaderEpoch int32
	// HighWatermark is the partition's high watermark at fetch time, so lag
	// can be computed against the position last processed. Zero for sources
	// without one.
	HighWatermark int64
}

Message is one record from a source, with whatever provenance the source knows about it. Only Kafka populates the metadata fields.

func (m Message) HasMetadata() bool

type MetadataWriter ¶

type MetadataWriter interface {
	WriteMessage(msg Message) error
}

MetadataWriter is implemented by handlers that can use a message's source metadata. Handlers that only need the payload implement Handler alone and the consume loop hands them the value.

type Metrics ¶

type Metrics struct {
	MessageCount           metric.Int64Counter
	HandlerRowsRead        metric.Int64Counter
	ErrorCount             metric.Int64Counter
	SourceReadLatency      metric.Float64Histogram
	SinkFlushLatency       metric.Float64Histogram
	SinkFlushNumRows       metric.Int64Gauge
	SinkFlushCount         metric.Int64Counter
	BatchProcessingLatency metric.Float64Histogram
	StateCommitLatency     metric.Float64Histogram
	StateCommitCount       metric.Int64Counter
	StateSizeBytes         metric.Int64Gauge
	StateTableRows         metric.Int64Gauge
	ReferenceTableRows     metric.Int64Gauge
	ConsumerLag            metric.Int64Gauge
}

Metrics holds the instruments the pipeline records, mirroring the names, descriptions and units the Python engine exports.

func NewMetrics(mp metric.MeterProvider) (*Metrics, error)

type OffsetStat ¶ added in v1.0.5

type OffsetStat struct {
	Topic       string `json:"topic"`
	Partition   int32  `json:"partition"`
	Offset      int64  `json:"offset"`
	LeaderEpoch int32  `json:"leader_epoch"`
}

OffsetStat is one partition's durable position.

type OffsetStore ¶ added in v1.0.5

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

OffsetStore persists Marks to DuckDB. It holds no lock and starts no transaction of its own: Save issues writes on the connection it is given, and it is the caller's job to commit -- typically together with the batch that advanced those offsets.

func NewOffsetStore(conn adbc.Connection) *OffsetStore

func (s *OffsetStore) Init(ctx context.Context) error

func (s *OffsetStore) Load(ctx context.Context) (*Marks, error)

func (s *OffsetStore) Save(ctx context.Context, marks *Marks) error

type PipelineErrorPolicies ¶

type PipelineErrorPolicies struct {
	Policy  ErrorPolicy
	DLQSink Sink
}

type RefTable ¶ added in v1.1.0

type RefTable struct {
	Catalog string
	Schema  string
	Name    string
}

RefTable is a table the handler SQL reads that the pipeline did not create.

func ReferenceTables(
	ctx context.Context,
	conn adbc.Connection,
	sql string,
	managed []string,
) ([]RefTable, error)

func (r RefTable) Attached() bool

func (r RefTable) Qualified() string

type Sink ¶

type Sink interface {
	WriteTable(ctx context.Context, batch arrow.Table) error
	Flush(ctx context.Context) error
}

Sink is where a batch leaves the pipeline.

Both write paths take a context so a sink that retries can be interrupted. Without it, a retry ladder outlives the SIGTERM that asked the pipeline to stop, and the graceful drain waits out the full retry deadline before it can finish. See #110.

func NewCountingSink(inner Sink, mp metric.MeterProvider, sinkType, role string) Sink

type Source ¶

type Source interface {
	Start() error
	Stream() <-chan []Message
	Commit() error
	Close() error
}

type StateStats ¶ added in v1.0.5

type StateStats struct {
	Path      string       `json:"path"`
	SizeBytes int64        `json:"size_bytes"`
	Tables    []TableStat  `json:"tables"`
	Offsets   []OffsetStat `json:"offsets"`
}

StateStats is a snapshot of a pipeline's durable state: how large it is, what is in it, and how far through the stream it has committed.

It is produced once and rendered by two consumers -- the /stats endpoint of a running pipeline and the CLI reading a stopped one -- so the two can never disagree about what a pipeline's state contains.

func CollectStateStats(ctx context.Context, conn adbc.Connection, path string) (*StateStats, error)

type Stats ¶

type Stats struct {
	StartTime time.Time
	NumErrors int
	// contains filtered or unexported fields
}

func (s *Stats) AddMessagesConsumed(n int64)

func (s *Stats) GetThroughput() float64

func (s *Stats) MessagesConsumed() int64

func (s *Stats) SetNumMessagesConsumed(num int64)

func (s *Stats) SetThroughput(throughput float64)

type TableStat ¶ added in v1.0.5

type TableStat struct {
	Name string `json:"name"`
	Rows int64  `json:"rows"`
}

TableStat is one managed table and how much is in it.

type Turbine ¶

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

func NewTurbine(
	source Source,
	handler Handler,
	sink Sink,
	batchSize int,
	flushInterval time.Duration,
	lock *sync.Mutex,
	policy PipelineErrorPolicies,
	opts ...TurbineOption,
) *Turbine

func (t *Turbine) ConsumeLoop(ctx context.Context, maxMsgs int) (stats *Stats, err error)

func (t *Turbine) StatusLoop(ctx context.Context) error

func (t *Turbine) SyncState(ctx context.Context) error

type TurbineOption ¶

type TurbineOption func(turbine *Turbine)

func WithMetrics(m *Metrics) TurbineOption

func WithStateStats(fn func() (*StateStats, error)) TurbineOption

func WithStateStore(offsets offsetSaver, tx stateTx) TurbineOption

func WithTurbineLogger(l *zap.Logger) TurbineOption