package columnar

import "github.com/krabka-io/krabka-streams-go/columnar"

Package columnar processes Kafka records as Apache Arrow batches instead of one record at a time.

It is a separate execution model from Kafka Streams: it does not build a Kafka topology, but its runner can participate in a consumer group and retain partition-local state. It is meant for analytical work where per-record dispatch dominates the cost.

One fetched topic-partition batch is one processing unit. Records arrive from a consumer poll, are decoded into a single arrow.Record, flow through operators as whole batches, and are encoded back into records at a sink.

Arrow memory is reference-counted. Whatever a public method returns to you, you release; batches inside a running topology belong to the framework.

Index

Constants

const (
    // KeyColumn holds the record key bytes; null for a keyless record.
    KeyColumn = "__key"

    // TimestampColumn holds the record timestamp as a signed 64-bit integer.
    TimestampColumn = "__timestamp"

    // PartitionColumn holds the source partition as a signed 32-bit integer.
    PartitionColumn = "__partition"

    // OffsetColumn holds the source offset as a signed 64-bit integer.
    OffsetColumn = "__offset"

    // HeadersColumn holds the ordered Kafka headers, including null values.
    HeadersColumn = "__headers"
)

Reserved metadata column names. Every decoded batch carries these five columns, appended after the payload columns in this order. 

const BarrierStateTopic = "__barrier_state"

BarrierStateTopic is the internal topic a krabka broker writes the barrier group definitions, the injection starts, and the cuts to. Read it with a CutReader. 

const DefaultMaxRecordBytes = 900 * 1024

DefaultMaxRecordBytes is the soft cap on one encoded record, chosen to stay under a 1 MiB broker message limit with room for headers and framing. 

const DefaultMaxUncompressedBytes = 16 * 1024 * 1024

DefaultMaxUncompressedBytes is the decompression ceiling of a GzipBatchCodec. 

const NoEpoch int64 = -1

NoEpoch keys the snapshot a runner saves outside a barrier, such as the one it writes when a rebalance revokes a partition. A barrier keys its snapshot by the epoch of the cut instead. 

const WindowEndColumn = "__window_end"

WindowEndColumn is the epoch-millisecond window end added by windowed grouping. 

const WindowStartColumn = "__window_start"

WindowStartColumn is the epoch-millisecond window start added by windowed grouping.

Functions

func AppendValue source

func AppendValue(builder array.Builder, value any) error

AppendValue appends one Go value to an Arrow builder, coercing it to the declared Arrow type; nil appends a null.

String columns accept anything by formatting it; binary columns accept bytes and strings; integer columns accept integral values checked for overflow; unsigned columns additionally reject negatives; date and timestamp columns accept integers or time.Time; time-of-day columns accept integers or time.Duration; decimal columns accept *big.Rat, *big.Int, numbers, and numeric text; list, struct, and map columns accept slices and map[string]any; union columns pick the first member that accepts the value.

func PayloadColumn source

func PayloadColumn(name string) string

PayloadColumn returns the escaped name a payload column receives in the processing batch when its name collides with a reserved metadata column. Non-colliding names are returned unchanged.

func RunGroupOnce source

func RunGroupOnce(ctx context.Context, topology *BuiltTopology, consumer Consumer, producer Producer, pollTimeout time.Duration) (map[TopicPartition]int64, error)

RunGroupOnce polls a subscribed consumer once, runs every fetched partition, sends the outputs, and commits the consumed offsets. It returns the committed next offsets by partition.

func RunPartitionOnce source

func RunPartitionOnce(ctx context.Context, topology *BuiltTopology, consumer Consumer, producer Producer, topic string, partition int, offset int64, pollTimeout time.Duration) (int64, error)

RunPartitionOnce runs one explicit single-partition cycle: assign, seek, poll once, run the topology, send, wait for acknowledgements, commit.

It returns offset unchanged when the poll was empty, without producing anything, and otherwise the highest consumed offset plus one. Failed processor state is rolled back before the error is returned.

func SendAll source

func SendAll(outputs []ProducedToTopic, producer Producer) error

SendAll dispatches every produced record and waits for all acknowledgements, returning the first send error.

func Subscribe source

func Subscribe(topology *Topology, consumer Consumer) error

Subscribe subscribes a consumer to every source topic of a topology without a rebalance listener.

func Value source

func Value(arr arrow.Array, row int) any

Value reads one value from an Arrow array as a native Go value.

Null values are nil; strings, byte slices, sized integers, floats, and booleans map to their Go counterparts; decimals become *big.Rat; lists become []any; structs and maps become map[string]any; unions yield the active member's value. Dates, times, and timestamps are returned as their raw integer representation.

Types

type AggregateFunction source

type AggregateFunction int

AggregateFunction enumerates the built-in aggregations. 

const (
    // Count counts rows in the group, nulls included. The output type is a
    // signed 64-bit integer.
    Count AggregateFunction = iota

    // Sum accumulates exactly for integral inputs and as float64 for
    // floating-point inputs; an all-null group yields null.
    Sum

    // Min keeps the smallest non-null value.
    Min

    // Max keeps the largest non-null value.
    Max
)

type Aggregation source

type Aggregation struct {
    // InputColumn is the column the aggregate reads.
    InputColumn string

    // OutputColumn is the column the aggregate writes.
    OutputColumn string

    // Function is the aggregate function.
    Function AggregateFunction

    // OutputType overrides the output Arrow type when non-nil.
    OutputType arrow.DataType
}

Aggregation names an input column, an output column, and the function that folds one into the other. OutputType overrides the inferred output type when non-nil.

type Barrier source

type Barrier struct {
    // Cut is the cut the runner aligned on.
    Cut BarrierCut

    // Partitions are the logical partition numbers whose state the runner
    // snapshotted, in ascending order.
    Partitions []int

    // Offsets are the committed offsets of the aligned topic partitions.
    // Each one is the marker offset of the cut.
    Offsets map[TopicPartition]int64
}

Barrier is the barrier a group runner reached: the cut it aligned on, the partitions whose state it snapshotted, and the offsets it committed.

type BarrierCut source

type BarrierCut struct {
    // Group is the barrier group the cut belongs to.
    Group string

    // Epoch is the epoch of the cut. A coordinator never reuses one.
    Epoch int64

    // TriggeredAt is the time the injection started, in epoch milliseconds.
    TriggeredAt int64

    // CompletedAt is the time the injection finished, in epoch milliseconds.
    CompletedAt int64

    // Status reports whether every partition received the marker.
    Status CutStatus

    // Offsets is the marker offset of every partition that received one.
    Offsets map[TopicPartition]int64

    // Missing names the partitions that received no marker. It is empty for
    // a complete cut.
    Missing []TopicPartition
}

BarrierCut is the manifest of one epoch of a barrier group: the offset of that epoch's marker in every partition of the group.

The marker itself is a Kafka control record. No consumer sees it, and the offset it holds never carries a data record, so the records before the cut of a partition are exactly the records with a lower offset.

func DecodeBarrierCut source

func DecodeBarrierCut(key, value []byte) (*BarrierCut, error)

DecodeBarrierCut decodes one record of BarrierStateTopic.

It returns a nil cut and a nil error for a record that carries no cut: a group definition, an injection start, and a tombstone. Malformed bytes return a *BarrierFormatError.

Example

A cut manifest on the internal barrier state topic names the offset of one epoch's marker in every partition of a barrier group. All integers are big-endian, and a string is an int16 byte length and then UTF-8 bytes. 

package main

import (
    "encoding/binary"
    "fmt"

    "github.com/krabka-io/krabka-streams-go/columnar"
)

func main() {
    key := binary.BigEndian.AppendUint16(nil, 0) // key version
    key = binary.BigEndian.AppendUint16(key, 2)  // kind: cut
    key = binary.BigEndian.AppendUint16(key, 5)  // group name length
    key = append(key, "audit"...)                // group name
    key = binary.BigEndian.AppendUint64(key, 7)  // epoch

    value := binary.BigEndian.AppendUint16(nil, 0)     // value version
    value = binary.BigEndian.AppendUint64(value, 1000) // triggered at
    value = binary.BigEndian.AppendUint64(value, 1200) // completed at
    value = append(value, 0)                           // status: complete
    value = binary.BigEndian.AppendUint32(value, 1)    // one topic
    value = binary.BigEndian.AppendUint16(value, 6)    // topic name length
    value = append(value, "orders"...)                 // topic name
    value = binary.BigEndian.AppendUint32(value, 1)    // one partition
    value = binary.BigEndian.AppendUint32(value, 0)    // partition 0
    value = binary.BigEndian.AppendUint64(value, 4200) // marker offset
    value = binary.BigEndian.AppendUint32(value, 0)    // no missing partition

    cut, err := columnar.DecodeBarrierCut(key, value)
    if err != nil {
        panic(err)
    }
    offset, _ := cut.Offset(columnar.TopicPartition{Topic: "orders", Partition: 0})
    fmt.Println(cut.Group, cut.Epoch, cut.Status, offset)
}

Output:

audit 7 complete 4200

func (BarrierCut) Complete source

func (c BarrierCut) Complete() bool

Complete reports whether every partition of the group received the marker. A runner aligns on a complete cut only.

func (BarrierCut) Offset source

func (c BarrierCut) Offset(partition TopicPartition) (int64, bool)

Offset returns the marker offset of one partition, and reports whether the cut names it.

type BarrierFormatError source

type BarrierFormatError struct {
    // Part names the record part that failed to decode, "key" or "cut
    // value".
    Part string
    // contains filtered or unexported fields
}

BarrierFormatError reports a malformed record of BarrierStateTopic.

func (*BarrierFormatError) Error source

func (e *BarrierFormatError) Error() string

Error describes the fault and names the record part that carries it.

type BarrierListener source

type BarrierListener func(Barrier)

BarrierListener runs after a group runner commits at a barrier. The snapshot of every partition is already in the state store, and the committed position of every aligned partition is the cut. The runner calls the listener on its own goroutine, so a slow listener slows the poll loop.

type BatchCodec source

type BatchCodec interface {
    // Decode turns a non-empty fetched batch into one Arrow batch.
    Decode(topic string, records []ConsumedRecord) (arrow.Record, error)

    // Encode turns an Arrow batch back into producible records.
    Encode(topic string, batch arrow.Record) ([]ProduceRecord, error)
}

BatchCodec converts between fetched Kafka records and Arrow batches.

Decode turns one fetched batch into one Arrow batch that carries the five reserved metadata columns; Encode turns an Arrow batch back into records. The caller owns the batch Decode returns and must release it. Encode returns plain byte-backed records and releases its internal temporaries itself.

type BlobCodec source

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

BlobCodec is the BatchCodec for records that are already Arrow: each Kafka record value holds an Arrow IPC stream with many rows.

Decoding reads each record value as an Arrow IPC stream, attaches metadata columns, and concatenates the results; every row keeps the metadata of the record it came from. All records in a batch must share one payload schema.

Encoding drops the metadata columns and serializes the payload as Arrow IPC records. It packs the largest consecutive rows that fit under the byte cap and share one key, timestamp, and header list, then applies that envelope to the output record. A single row that exceeds the cap fails instead of sending a record the broker will reject.

func NewBlobCodec source

func NewBlobCodec(mem memory.Allocator) *BlobCodec

NewBlobCodec creates a codec with DefaultMaxRecordBytes.

func NewBlobCodecWithMaxBytes source

func NewBlobCodecWithMaxBytes(mem memory.Allocator, maxRecordBytes int) (*BlobCodec, error)

NewBlobCodecWithMaxBytes creates a codec with a custom record byte cap. Raise the cap only alongside the broker's max.message.bytes.

func (*BlobCodec) Decode source

func (c *BlobCodec) Decode(topic string, records []ConsumedRecord) (arrow.Record, error)

Decode implements BatchCodec.

func (*BlobCodec) Encode source

func (c *BlobCodec) Encode(topic string, batch arrow.Record) ([]ProduceRecord, error)

Encode implements BatchCodec.

type BuiltTopology source

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

BuiltTopology is a validated, reusable columnar topology.

The processor instances created when a topology is built survive across calls to RunBatch, so built-in GroupBy accumulates across batches and custom processors can keep state. Processor instances are isolated by logical partition number, so co-partitioned source topics can join without sharing state with another partition.

A built topology serializes concurrent calls so Arrow batches are never shared across goroutines. All intermediate batches are released before RunBatch returns, including on the error path.

func (*BuiltTopology) Close source

func (b *BuiltTopology) Close() error

Close releases every partition.

func (*BuiltTopology) HasPartition source

func (b *BuiltTopology) HasPartition(partition int) bool

HasPartition reports whether the partition currently has processor state.

func (*BuiltTopology) ReleasePartition source

func (b *BuiltTopology) ReleasePartition(partition int)

ReleasePartition discards a partition's processor instances and state.

func (*BuiltTopology) RestorePartition source

func (b *BuiltTopology) RestorePartition(partition int, snapshots map[string][]byte) error

RestorePartition replaces the state of a partition's stateful processors with snapshots keyed by node name.

func (*BuiltTopology) RunBatch source

func (b *BuiltTopology) RunBatch(topic string, records []ConsumedRecord) ([]ProducedToTopic, error)

RunBatch evaluates the topology for one source topic's records and returns the produced records.

func (*BuiltTopology) RunBatches source

func (b *BuiltTopology) RunBatches(input map[string][]ConsumedRecord) ([]ProducedToTopic, error)

RunBatches evaluates several source topics together, which lets merges express fan-in across topics. Records are partitioned by their logical partition number, and each partition runs with its own processor state.

func (*BuiltTopology) RunPartitionBatches source

func (b *BuiltTopology) RunPartitionBatches(partition int, input map[string][]ConsumedRecord) ([]ProducedToTopic, error)

RunPartitionBatches evaluates one logical partition's records. Every record must belong to the given partition.

func (*BuiltTopology) SnapshotPartition source

func (b *BuiltTopology) SnapshotPartition(partition int) (map[string][]byte, error)

SnapshotPartition serializes every stateful processor of a partition, keyed by node name.

type BuiltinOp source

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

BuiltinOp is a built-in Processor that always forwards exactly one batch. Use Filter, Select, WithColumns, GroupBy, or WindowedGroupBy to create one.

func Filter source

func Filter(mem memory.Allocator, predicate RowPredicate) *BuiltinOp

Filter keeps the rows the predicate passes, in their original order, with the same schema. Metadata columns travel with the rows.

func GroupBy source

func GroupBy(mem memory.Allocator, keys []string, aggregations ...Aggregation) *BuiltinOp

GroupBy groups rows by the values of the key columns and retains those groups across every batch the same operator instance sees. The output is the current cumulative result, ordered by first appearance, with key columns first and aggregate columns after them. Metadata columns are dropped unless you group by them.

func Select source

func Select(mem memory.Allocator, columns ...string) *BuiltinOp

Select keeps the named payload columns, in the order given, then appends whichever reserved metadata columns exist in the input. Duplicate names are ignored after the first; a missing name fails.

func WindowedGroupBy source

func WindowedGroupBy(mem memory.Allocator, keys []string, windowSize time.Duration, aggregations ...Aggregation) (*BuiltinOp, error)

WindowedGroupBy is the same cumulative aggregation split into fixed event-time windows. It reads __timestamp and adds __window_start and __window_end as epoch milliseconds. Closed windows are retained for one window size.

func WindowedGroupByWithRetention source

func WindowedGroupByWithRetention(mem memory.Allocator, keys []string, windowSize, retention time.Duration, aggregations ...Aggregation) (*BuiltinOp, error)

WindowedGroupByWithRetention keeps closed windows for a retention duration no shorter than the window size.

func WithColumns source

func WithColumns(mem memory.Allocator, columns ...DerivedColumn) (*BuiltinOp, error)

WithColumns adds derived columns. A derived column whose name matches an existing column replaces it in place, keeping its position; a new name is appended after the existing columns. Reserved names are rejected here, at construction time.

func (*BuiltinOp) Process source

func (o *BuiltinOp) Process(ctx *Context, batch arrow.Record) error

Process implements Processor.

func (*BuiltinOp) Restore source

func (o *BuiltinOp) Restore(snapshot []byte) error

Restore implements StatefulProcessor.

func (*BuiltinOp) Snapshot source

func (o *BuiltinOp) Snapshot() ([]byte, error)

Snapshot implements StatefulProcessor. Stateless operators return nil.

type ConsumedRecord source

type ConsumedRecord struct {
    // Key is the record key, or nil for a keyless record.
    Key []byte

    // Value is the record value. A tombstone becomes an empty slice.
    Value []byte

    // Timestamp is the record timestamp in epoch milliseconds.
    Timestamp int64

    // Partition is the source partition.
    Partition int

    // Offset is the source offset.
    Offset int64

    // Headers are the ordered Kafka record headers.
    Headers []RecordHeader
}

ConsumedRecord is one record fetched from Kafka. Byte slices and header lists are defensively copied by NewConsumedRecord; consumer buffers get reused by Kafka clients, and Arrow-adjacent code holds references longer than a single callback.

func NewConsumedRecord source

func NewConsumedRecord(key, value []byte, timestamp int64, partition int, offset int64, headers ...RecordHeader) ConsumedRecord

NewConsumedRecord copies its arguments into an independent record.

type Consumer source

type Consumer interface {
    // Assign replaces the consumer's assignment with exactly these
    // partitions.
    Assign(partitions []TopicPartition) error

    // Seek positions the next fetch of a partition at an offset.
    Seek(partition TopicPartition, offset int64) error

    // Poll fetches records, waiting up to the timeout.
    Poll(ctx context.Context, timeout time.Duration) (map[TopicPartition][]ConsumedRecord, error)

    // CommitOffsets synchronously commits the given next-record offsets.
    CommitOffsets(ctx context.Context, offsets map[TopicPartition]int64) error

    // Subscribe joins the consumer group for the topics, delivering
    // assignment changes to the listener.
    Subscribe(topics []string, listener RebalanceListener) error

    // GroupMetadata returns the opaque consumer group metadata handed to
    // transactional offset commits.
    GroupMetadata() any
}

Consumer is the minimal Kafka consumer surface the runner needs. Adapters over franz-go or confluent-kafka-go implement it.

Poll returns the fetched records grouped by topic partition, in offset order within each partition. A record with a nil value is normalized by the runner to an empty value.

type Context source

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

Context collects the batches a processor forwards.

func (*Context) Forward source

func (c *Context) Forward(batch arrow.Record)

Forward emits one output batch to every downstream node. Calling it zero times drops the input; calling it several times fans the input out.

type CutReader source

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

CutReader reads the cuts of BarrierStateTopic with a manual assign, seek, and poll loop. It joins no consumer group.

Give it a consumer of its own. Assign replaces the assignment of the consumer it drives, which revokes the partitions of a subscribed group runner.

The reader keeps every cut it read and resumes where the previous call stopped, so only the first call reads the whole topic. It is safe for concurrent use.

func NewCutReader source

func NewCutReader(consumer Consumer, partitions int, options ...CutReaderOption) (*CutReader, error)

NewCutReader creates a reader over the partitions of the barrier state topic. partitions is the partition count of that topic.

func (*CutReader) CompleteCutsAfter source

func (r *CutReader) CompleteCutsAfter(ctx context.Context, group string, epoch int64) ([]BarrierCut, error)

CompleteCutsAfter returns the complete cuts of a barrier group with an epoch above epoch, oldest first. Partial cuts are left out: a partial cut has partitions that never receive the marker, so a task that waits for one waits forever.

func (*CutReader) LatestCompleteCut source

func (r *CutReader) LatestCompleteCut(ctx context.Context, group string) (*BarrierCut, error)

LatestCompleteCut returns the complete cut of a barrier group with the highest epoch. It returns a nil cut when the topic holds no complete cut of that group.

type CutReaderOption source

type CutReaderOption func(*CutReader)

CutReaderOption configures a CutReader.

func WithCutIdlePolls source

func WithCutIdlePolls(polls int) CutReaderOption

WithCutIdlePolls sets how many empty polls in a row end a read; the default is one.

func WithCutPollTimeout source

func WithCutPollTimeout(timeout time.Duration) CutReaderOption

WithCutPollTimeout sets the timeout of one poll; the default is 250 milliseconds.

func WithCutTopic source

func WithCutTopic(topic string) CutReaderOption

WithCutTopic reads the cuts from another topic than BarrierStateTopic.

type CutStatus source

type CutStatus int8

CutStatus reports whether every partition of a barrier group received the marker of an epoch. 

const (
    // CutComplete marks a cut that names a marker offset for every partition
    // of the group. Only a complete cut is alignable.
    CutComplete CutStatus = 0

    // CutPartial marks a cut that misses at least one partition. The
    // partitions in [BarrierCut.Missing] never receive the marker of that
    // epoch, so a task that waits for one waits forever. Skip a partial cut.
    CutPartial CutStatus = 1
)

func (CutStatus) String source

func (s CutStatus) String() string

String names the status.

type DerivedColumn source

type DerivedColumn struct {
    // Field declares the column name and Arrow type.
    Field arrow.Field

    // Value computes the column value for each row.
    Value RowValue
}

DerivedColumn pairs an Arrow field with the expression that fills it. Returned values are coerced to the declared Arrow type; nil writes a null.

type ErrorPolicy source

type ErrorPolicy struct {
    // Action is the policy action.
    Action ErrorPolicyAction

    // DeadLetterTopic is the dead-letter topic; required for
    // [ActionDeadLetter] and empty otherwise.
    DeadLetterTopic string
}

ErrorPolicy decides what a group runner does when a partition's batch fails. Failed processor state is rolled back before the policy is applied, and retriable errors always fail the poll so the records are retried.

func DeadLetterPolicy source

func DeadLetterPolicy(topic string) ErrorPolicy

DeadLetterPolicy forwards failed batches' records to a dead-letter topic with error-describing headers.

func FailPolicy source

func FailPolicy() ErrorPolicy

FailPolicy stops on the first failed batch.

func SkipPolicy source

func SkipPolicy() ErrorPolicy

SkipPolicy drops failed batches.

type ErrorPolicyAction source

type ErrorPolicyAction int

ErrorPolicyAction selects what a group runner does with a failed batch. 

const (
    // ActionFail stops processing and surfaces the error.
    ActionFail ErrorPolicyAction = iota

    // ActionSkip drops the failed batch and continues.
    ActionSkip

    // ActionDeadLetter forwards the failed batch's records to a dead-letter
    // topic and continues.
    ActionDeadLetter
)

type FileStateStore source

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

FileStateStore atomically saves partition snapshots to files in a directory, one file per partition and epoch.

func NewFileStateStore source

func NewFileStateStore(directory string) *FileStateStore

NewFileStateStore creates a store rooted at directory, which is created on the first save.

func (*FileStateStore) Load source

func (s *FileStateStore) Load(partition int, epoch int64) (map[string][]byte, error)

Load implements StateStore.

func (*FileStateStore) Save source

func (s *FileStateStore) Save(partition int, epoch int64, snapshot map[string][]byte) error

Save implements StateStore. The snapshot is written to a temporary file and atomically renamed over the target.

type GroupRunner source

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

GroupRunner is a reusable consumer-group runner. It subscribes to every source topic, retains one built topology across polls, and implements RebalanceListener so partition state follows assignments: newly assigned logical partitions restore from the state store, revoked partitions save before release, and lost partitions are released without saving.

func NewGroupRunner source

func NewGroupRunner(topology *Topology, consumer Consumer, producer Producer, options ...GroupRunnerOption) (*GroupRunner, error)

NewGroupRunner builds the topology, subscribes the consumer to its source topics with the runner as the rebalance listener, and returns the runner.

func (*GroupRunner) Close source

func (r *GroupRunner) Close() error

Close saves and releases every owned partition, then closes the topology.

func (*GroupRunner) Metrics source

func (r *GroupRunner) Metrics() *Metrics

Metrics returns the runner's metrics recorder.

func (*GroupRunner) OnPartitionsAssigned source

func (r *GroupRunner) OnPartitionsAssigned(partitions []TopicPartition)

OnPartitionsAssigned implements RebalanceListener.

func (*GroupRunner) OnPartitionsLost source

func (r *GroupRunner) OnPartitionsLost(partitions []TopicPartition)

OnPartitionsLost implements RebalanceListener; state is released without saving.

func (*GroupRunner) OnPartitionsRevoked source

func (r *GroupRunner) OnPartitionsRevoked(partitions []TopicPartition)

OnPartitionsRevoked implements RebalanceListener; state is saved before release.

func (*GroupRunner) RestoreToEpoch source

func (r *GroupRunner) RestoreToEpoch(ctx context.Context, epoch int64) (*BarrierCut, error)

RestoreToEpoch rewinds the runner to the cut of one epoch: it restores the state of every partition from that epoch's snapshot and seeks every input partition to its marker offset. It returns the cut it restored to.

The runner acts on the partitions it owns, or on every partition of the cut when it has no assignment yet.

func (*GroupRunner) RestoreToLatestCut source

func (r *GroupRunner) RestoreToLatestCut(ctx context.Context) (*BarrierCut, error)

RestoreToLatestCut rewinds the runner to the complete cut of the barrier group with the highest epoch, as GroupRunner.RestoreToEpoch does for a named epoch.

func (*GroupRunner) RunOnce source

func (r *GroupRunner) RunOnce(ctx context.Context, pollTimeout time.Duration) (map[TopicPartition]int64, error)

RunOnce polls once, processes, sends, and commits. With a barrier group it also holds the records after the pending cut, and snapshots the state of every partition once the poll reaches that cut.

func (*GroupRunner) RunOnceTransactional source

func (r *GroupRunner) RunOnceTransactional(ctx context.Context, pollTimeout time.Duration) (map[TopicPartition]int64, error)

RunOnceTransactional sends produced records and consumed offsets in one producer transaction. The producer must implement TransactionalProducer. A barrier commit rides inside that transaction.

func (*GroupRunner) Topology source

func (r *GroupRunner) Topology() *BuiltTopology

Topology returns the built topology the runner drives.

type GroupRunnerOption source

type GroupRunnerOption func(*GroupRunner)

GroupRunnerOption configures a GroupRunner.

func WithBarrierGroup source

func WithBarrierGroup(group string, reader *CutReader) GroupRunnerOption

WithBarrierGroup aligns the runner on the cuts of a barrier group. The reader must drive a consumer of its own, not the one the runner subscribes.

The runner then holds the records after each cut, snapshots the state of every partition at the cut, and commits the cut offsets.

func WithBarrierListener source

func WithBarrierListener(listener BarrierListener) GroupRunnerOption

WithBarrierListener sets the callback the runner calls after each barrier commit. It requires WithBarrierGroup.

func WithErrorPolicy source

func WithErrorPolicy(policy ErrorPolicy) GroupRunnerOption

WithErrorPolicy sets the error policy; the default fails fast.

func WithMetrics source

func WithMetrics(metrics *Metrics) GroupRunnerOption

WithMetrics sets the metrics recorder.

func WithStateStore source

func WithStateStore(store StateStore) GroupRunnerOption

WithStateStore sets the partition state store; the default is ephemeral.

type GzipBatchCodec source

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

GzipBatchCodec wraps any BatchCodec with per-record GZIP compression. Decoding decompresses each record value before delegating; encoding compresses each produced record value after delegating.

func NewGzipBatchCodec source

func NewGzipBatchCodec(delegate BatchCodec) *GzipBatchCodec

NewGzipBatchCodec wraps a codec with DefaultMaxUncompressedBytes.

func NewGzipBatchCodecWithMaxBytes source

func NewGzipBatchCodecWithMaxBytes(delegate BatchCodec, maxUncompressedBytes int) (*GzipBatchCodec, error)

NewGzipBatchCodecWithMaxBytes wraps a codec with a custom decompression ceiling.

func (*GzipBatchCodec) Decode source

func (c *GzipBatchCodec) Decode(topic string, records []ConsumedRecord) (arrow.Record, error)

Decode implements BatchCodec.

func (*GzipBatchCodec) Encode source

func (c *GzipBatchCodec) Encode(topic string, batch arrow.Record) ([]ProduceRecord, error)

Encode implements BatchCodec.

type IPCSerde source

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

IPCSerde reads and writes single Arrow batches in the Arrow IPC stream format. BlobCodec uses it internally, and you can use it directly to read or write Arrow-valued topics.

Each serialized value contains exactly one record batch. Deserialization reads the first batch and returns a batch owned by the caller: the caller must release it. Nil bytes deserialize to nil, and a nil batch serializes to nil, so Kafka tombstones survive untouched.

func NewIPCSerde source

func NewIPCSerde(mem memory.Allocator) *IPCSerde

NewIPCSerde creates a serde whose deserialized batches are allocated from mem.

func (*IPCSerde) Deserialize source

func (s *IPCSerde) Deserialize(topic string, data []byte) (arrow.Record, error)

Deserialize reads the first record batch of an Arrow IPC stream. The caller owns the returned batch and must release it.

func (*IPCSerde) Serialize source

func (s *IPCSerde) Serialize(topic string, batch arrow.Record) ([]byte, error)

Serialize writes one batch as an Arrow IPC stream.

type JSONRowBridge source

type JSONRowBridge[T any] struct {
    // contains filtered or unexported fields
}

JSONRowBridge is a RowBridge that routes values through JSON.

Column inference uses the first non-null sample and is retained by the bridge for every later batch; to pin the schema up front, use NewJSONRowBridgeWithSchema or JSONRowBridgeFromJSONSchema. Nested objects and arrays become serialized JSON text columns tagged with the krabka.json field metadata, so they survive a round trip.

Scalar row types (strings, numbers, booleans, and slices) have no fields to spread across columns, so they are wrapped in a single column named "value". Column order follows first appearance across the rows.

func JSONRowBridgeFromJSONSchema source

func JSONRowBridgeFromJSONSchema[T any](jsonSchema string) (*JSONRowBridge[T], error)

JSONRowBridgeFromJSONSchema derives the Arrow fields from a JSON Schema's properties, required list, local $ref references, primitive types, arrays, objects, and base64 content. Required fields reject nulls before a batch can escape.

func NewJSONRowBridge source

func NewJSONRowBridge[T any]() *JSONRowBridge[T]

NewJSONRowBridge creates a bridge that infers its Arrow schema from the first batch.

func NewJSONRowBridgeWithSchema source

func NewJSONRowBridgeWithSchema[T any](schema *arrow.Schema) *JSONRowBridge[T]

NewJSONRowBridgeWithSchema creates a bridge with a pinned Arrow schema.

func (*JSONRowBridge[T]) BatchToRows source

func (b *JSONRowBridge[T]) BatchToRows(batch arrow.Record) ([]T, error)

BatchToRows implements RowBridge.

func (*JSONRowBridge[T]) RowsToBatch source

func (b *JSONRowBridge[T]) RowsToBatch(rows []T, mem memory.Allocator) (arrow.Record, error)

RowsToBatch implements RowBridge.

type Join source

type Join struct {
    // LeftKey is the join key column on the left side.
    LeftKey string

    // RightKey is the join key column on the right side.
    RightKey string

    // Window is the event-time window; rows whose timestamps differ by more
    // than the window never join.
    Window time.Duration

    // LeftPrefix prefixes left payload columns in the output. Empty means
    // "left_".
    LeftPrefix string

    // RightPrefix prefixes right payload columns in the output. Empty means
    // "right_".
    RightPrefix string
}

Join configures a stateful co-partitioned inner equi-join within an event-time window.

type Metrics source

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

Metrics counts the batches, records, failures, and processing time of a runner. All methods are safe for concurrent use.

func NewMetrics source

func NewMetrics() *Metrics

NewMetrics creates a zeroed metrics recorder.

func (*Metrics) Snapshot source

func (m *Metrics) Snapshot() MetricsSnapshot

Snapshot reads the current totals.

type MetricsSnapshot source

type MetricsSnapshot struct {
    // Batches is the number of partition batches processed, including failed
    // ones.
    Batches int64

    // InputRecords is the number of consumed records seen.
    InputRecords int64

    // OutputRecords is the number of records produced by successful batches.
    OutputRecords int64

    // Failures is the number of failed partition batches.
    Failures int64

    // DeadLetterRecords is the number of records forwarded to a dead-letter
    // topic.
    DeadLetterRecords int64

    // ProcessingNanos is the total processing time in nanoseconds.
    ProcessingNanos int64
}

MetricsSnapshot is a point-in-time reading of a Metrics.

type Node source

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

Node is an opaque handle to a topology node, passed as the parent of later nodes.

type PositionReader source

type PositionReader interface {
    Positions(ctx context.Context, partitions []TopicPartition) (map[TopicPartition]int64, error)
}

PositionReader reports the next offset of assigned partitions. A consumer adapter implements it so an idle partition can reach a barrier without manufacturing a record.

type Processor source

type Processor interface {
    // Process handles one batch.
    Process(ctx *Context, batch arrow.Record) error
}

Processor transforms one Arrow batch into zero or more Arrow batches by calling Context.Forward once per output batch.

The input batch belongs to the framework: never release it. Forwarding a batch transfers its ownership to the framework; a batch you create and do not forward is yours to release.

type ProcessorFunc source

type ProcessorFunc func(ctx *Context, batch arrow.Record) error

ProcessorFunc adapts a function to the Processor interface.

func (ProcessorFunc) Process source

func (f ProcessorFunc) Process(ctx *Context, batch arrow.Record) error

Process implements Processor.

type ProduceRecord source

type ProduceRecord struct {
    // Key is the record key, or nil for a keyless record.
    Key []byte

    // Value is the record value.
    Value []byte

    // Timestamp is the record timestamp in epoch milliseconds. A negative
    // timestamp lets the producer apply its own.
    Timestamp int64

    // Headers are the ordered Kafka record headers.
    Headers []RecordHeader
}

ProduceRecord is one record to be written to Kafka. Byte slices and header lists are defensively copied by NewProduceRecord.

func NewProduceRecord source

func NewProduceRecord(key, value []byte, timestamp int64, headers ...RecordHeader) ProduceRecord

NewProduceRecord copies its arguments into an independent record.

func (ProduceRecord) Equal source

func (r ProduceRecord) Equal(other ProduceRecord) bool

Equal reports whether two produce records carry the same key, value, timestamp, and headers.

type ProducedToTopic source

type ProducedToTopic struct {
    // Topic is the sink topic.
    Topic string

    // Record is the produced record.
    Record ProduceRecord
}

ProducedToTopic pairs a produced record with its sink topic.

type Producer source

type Producer interface {
    Send(topic string, record ProduceRecord, done func(error))
}

Producer is the minimal Kafka producer surface the runner needs. Send dispatches a record asynchronously and calls done exactly once with the acknowledgement outcome.

type RebalanceListener source

type RebalanceListener interface {
    // OnPartitionsAssigned is called after partitions are assigned.
    OnPartitionsAssigned(partitions []TopicPartition)

    // OnPartitionsRevoked is called before partitions are revoked
    // cooperatively.
    OnPartitionsRevoked(partitions []TopicPartition)

    // OnPartitionsLost is called when partitions were lost without a clean
    // revocation.
    OnPartitionsLost(partitions []TopicPartition)
}

RebalanceListener receives partition assignment changes from a consumer. GroupRunner implements it.

type RecordHeader source

type RecordHeader struct {
    // Key is the header key.
    Key string

    // Value is the header value, or nil for a null-valued header.
    Value []byte
}

RecordHeader is one ordered Kafka record header. Header values may be nil, and keys may repeat.

func (RecordHeader) Equal source

func (h RecordHeader) Equal(other RecordHeader) bool

Equal reports whether two headers have the same key and value bytes.

type RowBridge source

type RowBridge[T any] interface {
    // RowsToBatch converts rows into one Arrow batch.
    RowsToBatch(rows []T, mem memory.Allocator) (arrow.Record, error)

    // BatchToRows converts an Arrow batch back into typed rows.
    BatchToRows(batch arrow.Record) ([]T, error)
}

RowBridge converts between typed rows and Arrow batches. The caller owns the batch RowsToBatch returns and must release it.

type RowCodec source

type RowCodec[T any] struct {
    // contains filtered or unexported fields
}

RowCodec is the BatchCodec for ordinary Kafka records that hold one value each, giving a columnar view of them.

Decoding deserializes each record value with the supplied serde, converts the values into Arrow columns through a RowBridge, and attaches metadata. Encoding reverses it: the payload columns become typed rows again, each row becomes one record, the key comes from __key, and the timestamp from __timestamp (or 0). Row count is preserved in both directions.

func NewRowCodec source

func NewRowCodec[T any](valueSerde ValueSerde[T], bridge RowBridge[T], mem memory.Allocator) *RowCodec[T]

NewRowCodec creates a codec over a value serde and a row bridge.

func (*RowCodec[T]) Decode source

func (c *RowCodec[T]) Decode(topic string, records []ConsumedRecord) (arrow.Record, error)

Decode implements BatchCodec.

func (*RowCodec[T]) Encode source

func (c *RowCodec[T]) Encode(topic string, batch arrow.Record) ([]ProduceRecord, error)

Encode implements BatchCodec.

type RowPredicate source

type RowPredicate func(batch arrow.Record, row int) bool

RowPredicate decides whether a row passes a filter.

type RowValue source

type RowValue func(batch arrow.Record, row int) (any, error)

RowValue computes a derived value for a row.

type StateStore source

type StateStore interface {
    // Load returns the saved snapshot of a partition at an epoch, empty when
    // none exists.
    Load(partition int, epoch int64) (map[string][]byte, error)

    // Save persists the snapshot of a partition at an epoch.
    Save(partition int, epoch int64, snapshot map[string][]byte) error
}

StateStore persists partition snapshots across rebalances and restarts. The key of a snapshot is the partition and the epoch, so the snapshots a barrier group takes at its cuts stay apart from each other and from the NoEpoch snapshot of a rebalance.

func NoStateStore source

func NoStateStore() StateStore

NoStateStore returns a store that loads nothing and saves nothing, for intentionally ephemeral state.

type StatefulProcessor source

type StatefulProcessor interface {
    Processor

    // Snapshot serializes the processor state.
    Snapshot() ([]byte, error)

    // Restore replaces the processor state with a snapshot.
    Restore(snapshot []byte) error
}

StatefulProcessor is a Processor that participates in partition snapshot and restore.

type TopicPartition source

type TopicPartition struct {
    // Topic is the topic name.
    Topic string

    // Partition is the partition number.
    Partition int
}

TopicPartition identifies one partition of one topic.

type Topology source

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

Topology is a builder for a columnar processing graph. Nodes are added in order, each non-source node names a parent that must already exist, and Topology.Build validates the result. A topology is not safe for concurrent use while building.

func NewTopology source

func NewTopology(mem memory.Allocator) *Topology

NewTopology creates a builder whose batches are allocated from mem.

Example

A topology decodes fetched records into one Arrow batch, runs operators on whole batches, and encodes the survivors back into records at a sink. 

package main

import (
    "fmt"

    "github.com/apache/arrow-go/v18/arrow"
    "github.com/apache/arrow-go/v18/arrow/array"
    "github.com/apache/arrow-go/v18/arrow/memory"

    "github.com/krabka-io/krabka-streams-go/columnar"
)

// rawSerde passes record values through unchanged.
type rawSerde struct{}

func (rawSerde) Serialize(topic string, value string) ([]byte, error) {
    return []byte(value), nil
}

func (rawSerde) Deserialize(topic string, data []byte) (string, error) {
    return string(data), nil
}

func main() {
    mem := memory.NewGoAllocator()
    codec := columnar.NewRowCodec[string](rawSerde{}, columnar.NewJSONRowBridge[string](), mem)

    topology := columnar.NewTopology(mem)
    source, err := topology.AddSource("source", []string{"in"}, codec)
    if err != nil {
        panic(err)
    }
    keep, err := topology.AddOperator("keep", columnar.Filter(mem,
        func(batch arrow.Record, row int) bool {
            values := batch.Column(0).(*array.String)
            return values.Value(row) != "drop"
        }), source)
    if err != nil {
        panic(err)
    }
    if _, err := topology.AddSink("sink", "out", codec, keep); err != nil {
        panic(err)
    }
    built, err := topology.Build()
    if err != nil {
        panic(err)
    }

    produced, err := built.RunBatch("in", []columnar.ConsumedRecord{
        columnar.NewConsumedRecord(nil, []byte("keep-me"), 1, 0, 0),
        columnar.NewConsumedRecord(nil, []byte("drop"), 2, 0, 1),
    })
    if err != nil {
        panic(err)
    }
    for _, output := range produced {
        fmt.Println(output.Topic, string(output.Record.Value))
    }
}

Output:

out keep-me

func (*Topology) AddJoin source

func (t *Topology) AddJoin(name string, join Join, left, right Node) (Node, error)

AddJoin adds a stateful co-partitioned inner equi-join within an event-time window.

func (*Topology) AddMerge source

func (t *Topology) AddMerge(name string, parents []Node) (Node, error)

AddMerge concatenates two or more same-schema upstream branches.

func (*Topology) AddOperator source

func (t *Topology) AddOperator(name string, operator *BuiltinOp, parent Node) (Node, error)

AddOperator adds a built-in operator.

func (*Topology) AddPassThroughSink source

func (t *Topology) AddPassThroughSink(name, topic string, source Node) (Node, error)

AddPassThroughSink copies source records byte-for-byte to a topic without codec work. It must be attached directly to a source.

func (*Topology) AddProcessor source

func (t *Topology) AddProcessor(name string, processor func() Processor, parent Node) (Node, error)

AddProcessor adds a custom processor; its factory is invoked once per logical partition.

func (*Topology) AddSink source

func (t *Topology) AddSink(name, topic string, codec BatchCodec, parent Node) (Node, error)

AddSink encodes its parent's batches to a topic.

func (*Topology) AddSource source

func (t *Topology) AddSource(name string, topics []string, codec BatchCodec) (Node, error)

AddSource adds a source that decodes records for any of the topics. At least one topic is required.

func (*Topology) Build source

func (t *Topology) Build() (*BuiltTopology, error)

Build validates the topology and returns a reusable BuiltTopology.

func (*Topology) SourceTopics source

func (t *Topology) SourceTopics() []string

SourceTopics returns every topic named by any source, for subscribing a consumer.

func (*Topology) Validate source

func (t *Topology) Validate() error

Validate checks the graph: node names are unique, every non-source node has a parent that was added earlier, and at least one source and one sink exist. Because a parent must already exist when a child is added, cycles are impossible by construction.

type TransactionalProducer source

type TransactionalProducer interface {
    Producer

    // BeginTransaction starts a transaction.
    BeginTransaction() error

    // SendOffsets adds consumed offsets to the open transaction.
    SendOffsets(offsets map[TopicPartition]int64, groupMetadata any) error

    // CommitTransaction commits the open transaction.
    CommitTransaction() error

    // AbortTransaction aborts the open transaction.
    AbortTransaction() error
}

TransactionalProducer extends Producer with the transactional flow used by GroupRunner.RunOnceTransactional. Configure and initialize transactions before creating the runner.

type ValueSerde source

type ValueSerde[T any] interface {
    // Serialize encodes a value for a topic.
    Serialize(topic string, value T) ([]byte, error)

    // Deserialize decodes bytes read from a topic.
    Deserialize(topic string, data []byte) (T, error)
}

ValueSerde converts one record value to and from bytes. The serdes in the schema package satisfy this interface.