operator

type BarrierSnapshotter ¶ added in v0.4.0

type BarrierSnapshotter interface {
	// SetBarrierSnapshot registers a callback invoked with the
	// operator's state each time a barrier passes through it.
	SetBarrierSnapshot(fn func(checkpointID string, snapshot []byte, err error))
}

BarrierSnapshotter is implemented by stateful operators that can snapshot their state synchronously when a checkpoint barrier passes through their Process loop. Snapshotting there — between two records — is the only race-free point: the operator is quiescent and its state reflects exactly the pre-barrier stream. Snapshots taken from outside (e.g. when the barrier reaches the end of the pipeline) race with the operator processing post-barrier records.

type Cloneable ¶ added in v0.4.0

type Cloneable interface {
	Clone() Operator
}

Cloneable is an optional interface for operators that can be duplicated with independent state. Stateful operators (Window, Reduce) implement this so each keyed worker gets its own instance with an isolated state backend.

type DescribableOperator ¶ added in v0.3.0

type DescribableOperator interface {
	DescribeOp() OperatorMeta
}

DescribableOperator is an optional interface that operators implement to expose their configuration for the dashboard.

type FilterOperator ¶

type FilterOperator struct {
	Parallelizable
	Fn    func(types.Record) bool
	Label string
}

FilterOperator keeps records that match the predicate and drops the rest.

func Filter(fn func(types.Record) bool) *FilterOperator

func (op *FilterOperator) DescribeOp() OperatorMeta

func (op *FilterOperator) GetLabel() string

func (op *FilterOperator) Name() string

func (op *FilterOperator) Process(in <-chan types.Record, out chan<- types.Record)

func (op *FilterOperator) ProcessOne(r types.Record) []types.Record

type FlatMapOperator ¶

type FlatMapOperator struct {
	Parallelizable
	Fn    func(types.Record) []types.Record
	Label string
}

FlatMapOperator applies a 1:many transformation to each record. The function returns a slice of records. If the slice is empty, the input record is effectively filtered out.

func FlatMap(fn func(types.Record) []types.Record) *FlatMapOperator

func (op *FlatMapOperator) DescribeOp() OperatorMeta

func (op *FlatMapOperator) GetLabel() string

func (op *FlatMapOperator) Name() string

func (op *FlatMapOperator) Process(in <-chan types.Record, out chan<- types.Record)

func (op *FlatMapOperator) ProcessOne(r types.Record) []types.Record

type KeyByOperator ¶

type KeyByOperator struct {
	KeySelector KeySelector
	Partitions  int
	Label       string
}

KeyByOperator marks the start of a keyed stage. Records flow through a router that hash-dispatches each record to one of N worker channels. Each worker runs its own copy of the downstream stateful operators (Window, Reduce) with isolated state.

KeyBy is not a processing operator — Execute() detects it and builds the worker topology. Process() is a no-op.

func KeyBy(fn KeySelector) *KeyByOperator

func (op *KeyByOperator) DescribeOp() OperatorMeta

func (op *KeyByOperator) GetLabel() string

func (op *KeyByOperator) IsRouter() bool

func (op *KeyByOperator) Name() string

func (op *KeyByOperator) Process(in <-chan types.Record, out chan<- types.Record)

func (op *KeyByOperator) Route(r types.Record) int

func (op *KeyByOperator) RouteKey(key []byte) int

func (op *KeyByOperator) SelectKey(r types.Record) []byte

func (op *KeyByOperator) WithPartitions(n int) *KeyByOperator

type KeySelector ¶ added in v0.4.0

type KeySelector func(types.Record) []byte

KeySelector extracts a routing key from a record. Used by KeyBy to deterministically route records to keyed workers.

type Labeled ¶

type Labeled interface {
	GetLabel() string
}

Labeled operators carry a user-provided label for display in the dashboard. If no label is set, the dashboard falls back to Name().

type MapOperator ¶

type MapOperator struct {
	Parallelizable
	Fn    func(types.Record) types.Record
	Label string
}

MapOperator applies a 1:1 transformation to each record. Every input record produces exactly one output record. To conditionally drop records, use Filter before or after Map.

func Map(fn func(types.Record) types.Record) *MapOperator

func (op *MapOperator) DescribeOp() OperatorMeta

func (op *MapOperator) GetLabel() string

func (op *MapOperator) Name() string

func (op *MapOperator) Process(in <-chan types.Record, out chan<- types.Record)

func (op *MapOperator) ProcessOne(r types.Record) []types.Record

type NativeSnapshotter ¶ added in v0.4.0

type NativeSnapshotter interface {
	SetNativeSnapshot(fn func(checkpointID string) (snapshot []byte, err error))
	Backend() state.StateBackend
}

NativeSnapshotter is implemented by stateful operators whose barrier-time snapshot can be replaced by a native (backend-level) checkpoint. When the engine detects a state.Checkpointable backend, it injects a snapshot function that hard-links the backend's files into the checkpoint's state directory and returns a small state-ref marker instead of serializing all state — this is what makes checkpoint cost scale with changed data instead of total state. The function runs at the same race-free point as BarrierSnapshotter (inside Process, as the barrier passes).

type Operator ¶

type Operator interface {
	// Name returns the operator type name (e.g. "Map", "Filter", "Window").
	// Used by the dashboard to display the pipeline graph.
	Name() string

	Process(in <-chan types.Record, out chan<- types.Record)
}

Operator transforms an input stream into an output stream. Each operator reads from an input channel, applies a transformation, and writes to an output channel. The output channel must be closed when the operator is done processing.

type OperatorMeta ¶ added in v0.3.0

type OperatorMeta struct {
	Type   string            `json:"type"`
	Label  string            `json:"label,omitempty"`
	Config map[string]string `json:"config,omitempty"`
}

OperatorMeta describes the configuration of an operator for the dashboard.

type Parallel ¶ added in v0.4.0

type Parallel interface {
	Parallelism() int
	SetParallelism(n int)
}

Parallel is implemented by operators that can run with multiple workers. The planner groups consecutive operators with the same parallelism into one execution stage.

type Parallelizable ¶ added in v0.4.0

type Parallelizable struct {
	Par int
}

Parallelizable provides the Parallel implementation for stateless operators via embedding. Zero value means parallelism 1.

func (p *Parallelizable) Parallelism() int

func (p *Parallelizable) SetParallelism(n int)

type ProcFailurePolicy ¶ added in v0.4.0

type ProcFailurePolicy int

ProcFailurePolicy determines what happens when a Process function returns an error. Follows the same pattern as sink FailurePolicy.

const (
	ProcFailureDrop ProcFailurePolicy = iota
	ProcFailureDLQ
	ProcFailureFail
)

type ProcessOperator ¶ added in v0.4.0

type ProcessOperator struct {
	Parallelizable
	Fn            func(types.Record) (types.Record, error)
	Label         string
	FailurePolicy ProcFailurePolicy
	DLQ           RecordSink
}

ProcessOperator wraps a user function that may return an error. On error, the failure policy is applied: drop the record, forward it to a DLQ, or fail the pipeline.

Watermarks and barriers pass through unchanged.

func NewProcess(fn func(types.Record) (types.Record, error), opts ...ProcessOption) *ProcessOperator

func (op *ProcessOperator) DescribeOp() OperatorMeta

func (op *ProcessOperator) GetLabel() string

func (op *ProcessOperator) Name() string

func (op *ProcessOperator) Process(in <-chan types.Record, out chan<- types.Record)

func (op *ProcessOperator) ProcessOne(r types.Record) []types.Record

type ProcessOption ¶ added in v0.4.0

type ProcessOption func(*ProcessOperator)

ProcessOption configures a ProcessOperator.

func WithProcessDLQ(dlq RecordSink) ProcessOption

func WithProcessFailurePolicy(p ProcFailurePolicy) ProcessOption

func WithProcessLabel(l string) ProcessOption

type RecordSink ¶ added in v0.4.0

type RecordSink interface {
	Write(ctx context.Context, r types.Record) error
}

RecordSink is a minimal sink interface used to forward failed records to a dead-letter queue. Implementations include a KafkaSink, PostgresSink, or any custom handler.

type ReduceFn ¶

type ReduceFn func(accum []byte, curr types.Record) []byte

ReduceFn takes the current accumulator and the incoming record, and returns the new accumulator bytes. The accumulator is persisted per key.

On the first record for a given key, accum will be nil (no previous state). The function should return the initial accumulator based on the first record.

Example (count per key):

reduce := Reduce(func(accum []byte, curr types.Record) []byte {
    count := 0
    if accum != nil {
        count = int(binary.BigEndian.Uint64(accum))
    }
    count++
    buf := make([]byte, 8)
    binary.BigEndian.PutUint64(buf, uint64(count))
    return buf
})

type ReduceOperator ¶

type ReduceOperator struct {
	Fn ReduceFn

	Label string
	// contains filtered or unexported fields
}

ReduceOperator maintains per-key state and applies a reduce function to merge each incoming record into the accumulator.

It must be used after KeyBy — the key determines which accumulator to use. On every record:

1. Look up the ValueState for the record's key
2. If state exists, use it as accumulator; otherwise accum is nil
3. Call reduceFn(accum, record) to get the new accumulator
4. Save the new accumulator to state
5. Emit the updated accumulator downstream as a new Record

func Reduce(fn ReduceFn) *ReduceOperator

func (op *ReduceOperator) Backend() state.StateBackend

func (op *ReduceOperator) Clone() Operator

func (op *ReduceOperator) DescribeOp() OperatorMeta

func (op *ReduceOperator) GetLabel() string

func (op *ReduceOperator) Name() string

func (op *ReduceOperator) Process(in <-chan types.Record, out chan<- types.Record)

func (op *ReduceOperator) Restore(data []byte) error

func (op *ReduceOperator) SetBarrierSnapshot(fn func(checkpointID string, snapshot []byte, err error))

func (op *ReduceOperator) SetNativeSnapshot(fn func(checkpointID string) ([]byte, error))

func (op *ReduceOperator) SetStateBackend(b state.StateBackend)

func (op *ReduceOperator) Snapshot() ([]byte, error)

type SingleProcessor ¶ added in v0.4.0

type SingleProcessor interface {
	ProcessOne(r types.Record) []types.Record
}

SingleProcessor is implemented by stateless operators that can be invoked directly, one record at a time, without channels. The execution engine chains SingleProcessors inside a stage as plain function calls.

The returned slice length encodes the operator semantics: 0 = record dropped (Filter), 1 = transformed (Map), N = fan-out (FlatMap).

Barriers and watermarks are never passed to ProcessOne — the stage machinery forwards and aligns them itself.

type Snapshotable ¶

type Snapshotable interface {
	// Snapshot returns the operator's current state as opaque bytes.
	// The returned bytes must be sufficient to fully reconstruct the
	// operator's state via Restore.
	Snapshot() ([]byte, error)

	// Restore replaces the operator's internal state from the given bytes.
	// The bytes must have been produced by a prior call to Snapshot on the
	// same operator type. Restore is called before the pipeline starts.
	Restore(data []byte) error
}

Snapshotable operators can snapshot and restore their internal state. This is used for checkpointing — when a barrier passes through a stateful operator, its state is captured and can be restored on recovery.

type StateConfigurable ¶ added in v0.4.0

type StateConfigurable interface {
	SetStateBackend(b state.StateBackend)
}

StateConfigurable is implemented by stateful operators whose state backend can be injected by the engine. The planner assigns each operator instance a backend created from the environment's BackendFactory (WithStateBackend), keyed by the owner ID used in checkpoint data. Operators keep a self-created in-memory backend as the default when no factory is configured.

SetStateBackend is called during plan construction, strictly before the operator processes any record or has state restored into it.

type WindowOperator ¶

type WindowOperator struct {
	Assigner    window.WindowAssigner
	IdleTimeout time.Duration
	Label       string
	// contains filtered or unexported fields
}

WindowOperator buffers records into time-based windows and fires them when the watermark passes the window's end time.

How it works:

1. Data records are assigned to windows by the WindowAssigner and buffered in the injected state backend (Pebble when configured), so window contents live on disk instead of the Go heap.
2. Watermark records advance the current watermark.
3. When the watermark passes a window's end time, that window is "closed" — all its buffered records are emitted as a single result.
4. Late records (timestamp < current watermark) are dropped.
5. Checkpoint barriers snapshot state (natively via the backend when it supports hard-link checkpoints, else serialized to JSON).

Must be used after KeyBy in a keyed stream so each key gets its own set of windows.

State layout in the backend:

• records: ListState(windowRecordsNS), keyed by windowKey.String() ("<recordKey>/<start>/<end>"); each entry is one serialized record. Window bounds are recovered from the key, so the set of open windows is exactly Keys().
• watermark: ValueState(windowWatermarkNS)["wm"] — 8-byte UnixNano. A RAM working copy (currentWatermark) serves the per-record late-drop check without a backend read.

func Window(assigner window.WindowAssigner) *WindowOperator

func (op *WindowOperator) Backend() state.StateBackend

func (op *WindowOperator) Clone() Operator

func (op *WindowOperator) CurrentWatermark() time.Time

func (op *WindowOperator) DescribeOp() OperatorMeta

func (op *WindowOperator) GetLabel() string

func (op *WindowOperator) Name() string

func (op *WindowOperator) Process(in <-chan types.Record, out chan<- types.Record)

func (op *WindowOperator) Restore(data []byte) error

func (op *WindowOperator) SetBarrierSnapshot(fn func(checkpointID string, snapshot []byte, err error))

func (op *WindowOperator) SetNativeSnapshot(fn func(checkpointID string) ([]byte, error))

func (op *WindowOperator) SetStateBackend(b state.StateBackend)

func (op *WindowOperator) Snapshot() ([]byte, error)

func (op *WindowOperator) WithIdleTimeout(d time.Duration) *WindowOperator