state

type BackendFactory ¶ added in v0.4.0

type BackendFactory func(ownerID string) (StateBackend, error)

BackendFactory creates a StateBackend for one state owner. An owner is a stateful operator instance, identified by the same stable IDs the checkpoint format uses: "op-<i>" for a top-level operator, "worker-<idx>" for a keyed-worker clone. The engine calls the factory once per owner while building the execution plan, so each owner gets isolated state (per-worker isolation is what keeps keyed state and barrier-time snapshots consistent).

Backends that also implement io.Closer are closed by the engine when the pipeline finishes.

func InMemory() BackendFactory

func Pebble(dir string) BackendFactory

type Checkpointable ¶ added in v0.4.0

type Checkpointable interface {
	CheckpointTo(dir string) error
	RestoreFrom(dir string) error
}

Checkpointable is an optional interface for backends that can checkpoint without serializing all state. The engine calls CheckpointTo during a barrier-triggered snapshot to produce a portable copy of the current state directory (hard-linked), and RestoreFrom on recovery to rebuild the live DB from that copy. This keeps checkpoint metadata small regardless of state size.

type ListState ¶

type ListState interface {
	// SetKey scopes all subsequent calls to this key.
	SetKey(key string)

	// Append adds a value to the list for the current key.
	Append(value []byte)

	// GetAll returns all values for the current key, or nil if none exist.
	GetAll() [][]byte

	// Clear removes all values for the current key.
	Clear()

	// Keys returns every key in this namespace that currently holds at
	// least one entry. Order is unspecified. It is independent of the
	// key set via SetKey. Used by operators that must iterate all their
	// keyed lists — e.g. Window firing every window whose end has
	// passed the watermark. Returns the keys only (small); records
	// within a key are loaded lazily via SetKey + GetAll.
	Keys() []string
}

ListState holds an ordered list of values per key. Think of it as a map[string][][]byte — each key gets an append-only list.

Before calling Append/GetAll/Clear, you must call SetKey to scope the operation to the current record's key.

type MemoryBackend ¶

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

MemoryBackend is an in-memory StateBackend using maps protected by a mutex. Each key gets its own entry. State is lost when the process restarts.

func NewMemoryBackend() *MemoryBackend

func (m *MemoryBackend) ListState(name string) ListState

func (m *MemoryBackend) ValueState(name string) ValueState

type PebbleBackend ¶ added in v0.4.0

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

PebbleBackend implements StateBackend on a CockroachDB Pebble LSM. Each owner gets its own Pebble DB (one owner per stateful operator instance — per-worker isolation). The working DB is disposable; durability comes from checkpoints (writes are unsynced; the WAL is never fsynced on the hot path).

func OpenPebble(dir string) (*PebbleBackend, error)

func (p *PebbleBackend) CheckpointTo(dir string) error

func (p *PebbleBackend) Close() error

func (p *PebbleBackend) ListState(name string) ListState

func (p *PebbleBackend) RestoreFrom(dir string) error

func (p *PebbleBackend) ValueState(name string) ValueState

type StateBackend ¶

type StateBackend interface {
	ValueState(name string) ValueState
	ListState(name string) ListState
}

StateBackend persists keyed state for stateful operators. Each Reduce/Process operator gets its own StateBackend instance, so keys don't collide across different operators.

type ValueState ¶

type ValueState interface {
	// SetKey scopes all subsequent Get/Set/Clear calls to this key.
	SetKey(key string)

	// Get returns the stored value for the current key, or nil if none exists.
	Get() []byte

	// Set stores a value for the current key, overwriting any previous value.
	Set(value []byte)

	// Clear removes the value for the current key.
	Clear()

	// SnapshotAll returns a copy of all key-value pairs in this state namespace.
	// Used for checkpointing.
	SnapshotAll() map[string][]byte

	// RestoreAll replaces all key-value pairs in this state namespace.
	// Used for recovery from a checkpoint.
	RestoreAll(entries map[string][]byte) error
}

ValueState holds a single value per key. Think of it as a map[string][]byte — each key gets one value.

Before calling Get/Set/Clear, you must call SetKey to scope the operation to the current record's key.