From 96c07bab1086510a55a004e9a1b90c30bc723a5c Mon Sep 17 00:00:00 2001 From: sangtran Date: Sat, 29 Aug 2026 01:08:28 +0700 Subject: feat: init base draft support window --- core/hash.go | 19 +++++++ core/operator/filter.go | 36 +++++++++++++ core/operator/keyby.go | 32 ++++++++++++ core/operator/keyed_process.go | 81 +++++++++++++++++++++++++++++ core/operator/map.go | 53 +++++++++++++++++++ core/operator/operator.go | 16 ++++++ core/operator/stateful_map.go | 75 +++++++++++++++++++++++++++ core/operator/watermark.go | 61 ++++++++++++++++++++++ core/operator/window.go | 110 +++++++++++++++++++++++++++++++++++++++ core/record.go | 40 +++++++++++++++ core/state.go | 10 ++++ core/state_backend.go | 62 ++++++++++++++++++++++ core/timer.go | 113 +++++++++++++++++++++++++++++++++++++++++ core/window.go | 54 ++++++++++++++++++++ 14 files changed, 762 insertions(+) create mode 100644 core/hash.go create mode 100644 core/operator/filter.go create mode 100644 core/operator/keyby.go create mode 100644 core/operator/keyed_process.go create mode 100644 core/operator/map.go create mode 100644 core/operator/operator.go create mode 100644 core/operator/stateful_map.go create mode 100644 core/operator/watermark.go create mode 100644 core/operator/window.go create mode 100644 core/record.go create mode 100644 core/state.go create mode 100644 core/state_backend.go create mode 100644 core/timer.go create mode 100644 core/window.go (limited to 'core') diff --git a/core/hash.go b/core/hash.go new file mode 100644 index 0000000..af15546 --- /dev/null +++ b/core/hash.go @@ -0,0 +1,19 @@ +package core + +import "hash/fnv" + +const DefaultMaxParallelism = 128 + +func AssignToKeyGroup(key string, maxParallelism int) int { + + if maxParallelism <= 0 { + maxParallelism = DefaultMaxParallelism + } + + // hash + h := fnv.New32a() + h.Write([]byte(key)) + hash := h.Sum32() + + return int(hash % uint32(maxParallelism)) +} diff --git a/core/operator/filter.go b/core/operator/filter.go new file mode 100644 index 0000000..80bbb77 --- /dev/null +++ b/core/operator/filter.go @@ -0,0 +1,36 @@ +package operator + +import ( + "context" + "flux/core" +) + +type FilterFunc[IN any] func(in IN) bool + +type FilterOperator[IN any] struct { + userFunc FilterFunc[IN] +} + +func NewFilterOperator[IN any](userFunc FilterFunc[IN]) *FilterOperator[IN] { + return &FilterOperator[IN]{userFunc: userFunc} +} + +func (f *FilterOperator[IN]) Open(ctx context.Context) error { + return nil +} + +func (f *FilterOperator[IN]) Process(ctx context.Context, record core.StreamRecord[IN], out Emitter[IN]) error { + if record.Type != core.RecordData { + return out.Emit(ctx, record) + } + + if f.userFunc(record.Value) { + return out.Emit(ctx, record) + } + + return nil +} + +func (f *FilterOperator[IN]) Close(ctx context.Context) error { + return nil +} diff --git a/core/operator/keyby.go b/core/operator/keyby.go new file mode 100644 index 0000000..56ef34e --- /dev/null +++ b/core/operator/keyby.go @@ -0,0 +1,32 @@ +package operator + +import ( + "context" + "flux/core" +) + +type KeySelector[IN any] func(in IN) string +type KeyByOperator[IN any] struct { + selector KeySelector[IN] +} + +func NewKeyByOperator[IN any](selector KeySelector[IN]) *KeyByOperator[IN] { + return &KeyByOperator[IN]{selector: selector} +} + +func (k *KeyByOperator[IN]) Open(ctx context.Context) error { + return nil +} + +func (k *KeyByOperator[IN]) Process(ctx context.Context, record core.StreamRecord[IN], out Emitter[IN]) error { + if record.Type != core.RecordData { + return out.Emit(ctx, record) + } + + record.Key = k.selector(record.Value) + return out.Emit(ctx, record) +} + +func (k *KeyByOperator[IN]) Close(ctx context.Context) error { + return nil +} diff --git a/core/operator/keyed_process.go b/core/operator/keyed_process.go new file mode 100644 index 0000000..1d3c86c --- /dev/null +++ b/core/operator/keyed_process.go @@ -0,0 +1,81 @@ +package operator + +import ( + "context" + "flux/core" + "fmt" +) + +type KeyedProcesser[IN, OUT any] interface { + Open(ctx context.Context) error + ProcessElement(ctx context.Context, record core.StreamRecord[IN], out Emitter[OUT]) error + Close(ctx context.Context) error + OnTimer(ctx context.Context, timestamp int64, out Emitter[OUT]) error +} + +type KeyProcessOperator[IN, OUT any] struct { + userFunc KeyedProcesser[IN, OUT] + timeService core.InternalTimerService + backend core.StateBackend +} + +func NewKeyProcessOperator[IN, OUT any](userFunc KeyedProcesser[IN, OUT]) *KeyProcessOperator[IN, OUT] { + return &KeyProcessOperator[IN, OUT]{ + userFunc: userFunc, + timeService: core.NewTimerService(), + } +} + +func (k *KeyProcessOperator[IN, OUT]) Open(ctx context.Context) error { + // init backend + be, ok := core.ExtractStateBackend(ctx) + if !ok { + return fmt.Errorf("state backend not found") + } + + k.backend = be + // inject the timer to context for user use + ctx = core.InjectTimerService(ctx, k.timeService) + return k.userFunc.Open(ctx) + +} + +func (k *KeyProcessOperator[IN, OUT]) Process(ctx context.Context, record core.StreamRecord[IN], out Emitter[OUT]) error { + + switch record.Type { + case core.RecordData: + if err := k.backend.SetCurrentKey(record.Key); err != nil { + return err + } + + return k.userFunc.ProcessElement(ctx, record, out) + case core.RecordWatermark: + // advance the timer + triggerTimerQueue := k.timeService.AdvanceWatermark(record.Timestamp) + + for _, t := range triggerTimerQueue { + + if err := k.backend.SetCurrentKey(t.Key); err != nil { + return err + } + if err := k.userFunc.OnTimer(ctx, t.Timestamp, out); err != nil { + return err + } + } + return out.Emit(ctx, core.StreamRecord[OUT]{ + Type: core.RecordWatermark, + Timestamp: record.Timestamp, + }) + default: + return out.Emit(ctx, core.StreamRecord[OUT]{ + Timestamp: record.Timestamp, + BarrierID: record.BarrierID, + Key: record.Key, + }) + } + +} + +func (k *KeyProcessOperator[IN, OUT]) Close(ctx context.Context) error { + return k.userFunc.Close(ctx) +} diff --git a/core/operator/map.go b/core/operator/map.go new file mode 100644 index 0000000..2165c25 --- /dev/null +++ b/core/operator/map.go @@ -0,0 +1,53 @@ +package operator + +import ( + "context" + "flux/core" + "fmt" +) + +// MapFunc is business logic for user implement +type MapFunc[IN, OUT any] func(in IN) (OUT, error) + +type MapOperator[IN, OUT any] struct { + userFunc MapFunc[IN, OUT] +} + +func NewMapOperator[IN, OUT any](fn MapFunc[IN, OUT]) *MapOperator[IN, OUT] { + return &MapOperator[IN, OUT]{ + userFunc: fn, + } +} + +func (fo *MapOperator[IN, OUT]) Open(ctx context.Context) error { + return nil +} + +func (fo *MapOperator[IN, OUT]) Process(ctx context.Context, record core.StreamRecord[IN], out Emitter[OUT]) error { + if record.Type != core.RecordData { + // watermark and barrier + return out.Emit(ctx, core.StreamRecord[OUT]{ + Type: record.Type, + Key: record.Key, + Timestamp: record.Timestamp, + BarrierID: record.BarrierID, + }) + } + + outValue, err := fo.userFunc(record.Value) + + if err != nil { + return fmt.Errorf("cannot process value %v: %w", record.Value, err) + } + + return out.Emit(ctx, core.StreamRecord[OUT]{ + Type: core.RecordData, + Key: record.Key, + Timestamp: record.Timestamp, + Value: outValue, + }) +} + +func (fo *MapOperator[IN, OUT]) Close(ctx context.Context) error { + return nil +} diff --git a/core/operator/operator.go b/core/operator/operator.go new file mode 100644 index 0000000..0c588e5 --- /dev/null +++ b/core/operator/operator.go @@ -0,0 +1,16 @@ +package operator + +import ( + "context" + "flux/core" +) + +type Emitter[OUT any] interface { + Emit(ctx context.Context, record core.StreamRecord[OUT]) error +} + +type Operator[IN, OUT any] interface { + Open(ctx context.Context) error + Process(ctx context.Context, record core.StreamRecord[IN], out Emitter[OUT]) error + Close(ctx context.Context) error +} diff --git a/core/operator/stateful_map.go b/core/operator/stateful_map.go new file mode 100644 index 0000000..295c656 --- /dev/null +++ b/core/operator/stateful_map.go @@ -0,0 +1,75 @@ +package operator + +import ( + "context" + "fmt" + + "flux/core" +) + +// RichMapFunc is a stateful map on the user side: Open grabs the state handles +// off ctx, Map runs per record with the backend already pointed at that +// record's key. +type RichMapFunc[IN, OUT any] interface { + Open(ctx context.Context) error + Map(ctx context.Context, in IN) (OUT, error) + Close(ctx context.Context) error +} + +type StatefulMapOperator[IN, OUT any] struct { + userFunc RichMapFunc[IN, OUT] + backend core.StateBackend +} + +func NewStatefulMapOperator[IN, OUT any](fn RichMapFunc[IN, OUT]) *StatefulMapOperator[IN, OUT] { + return &StatefulMapOperator[IN, OUT]{userFunc: fn} +} + +func (s *StatefulMapOperator[IN, OUT]) Open(ctx context.Context) error { + // Get backend state from context via using context KV + be, ok := core.ExtractStateBackend(ctx) + if !ok || be == nil { + return fmt.Errorf("stateful map: no state backend in context") + } + + s.backend = be + // The user func opens its ValueState off the same ctx. + return s.userFunc.Open(ctx) +} + +func (s *StatefulMapOperator[IN, OUT]) Process(ctx context.Context, record core.StreamRecord[IN], out Emitter[OUT]) error { + if record.Type != core.RecordData { + return out.Emit(ctx, core.StreamRecord[OUT]{ + Type: record.Type, + Timestamp: record.Timestamp, + Key: record.Key, + BarrierID: record.BarrierID, + }) + } + + // Without a key every record would silently share the "" state slot. + if record.Key == "" { + return fmt.Errorf("stateful map: record %v has no key, keyBy it first", record.Value) + } + + if err := s.backend.SetCurrentKey(record.Key); err != nil { + return fmt.Errorf("set current key %q: %w", record.Key, err) + } + + // run the user transform + res, err := s.userFunc.Map(ctx, record.Value) + if err != nil { + return fmt.Errorf("cannot process value %v: %w", record.Value, err) + } + + return out.Emit(ctx, core.StreamRecord[OUT]{ + Type: core.RecordData, + Timestamp: record.Timestamp, + Key: record.Key, + Value: res, + }) +} + +func (s *StatefulMapOperator[IN, OUT]) Close(ctx context.Context) error { + return s.userFunc.Close(ctx) +} diff --git a/core/operator/watermark.go b/core/operator/watermark.go new file mode 100644 index 0000000..3d591e5 --- /dev/null +++ b/core/operator/watermark.go @@ -0,0 +1,61 @@ +package operator + +import ( + "context" + + "flux/core" +) + +// TimestampAssigner is responsible for event time defined by user +type TimestampAssigner[IN any] func(in IN) int64 + +type WaterGeneratorOperator[IN any] struct { + assigner TimestampAssigner[IN] + maxOutOfOrder int64 + currentMaxTimestamp int64 + lastEmittedWatermark int64 +} + +func NewWatermarkGeneratorOperator[IN any](maxOutOfOrder int64, assigner TimestampAssigner[IN]) *WaterGeneratorOperator[IN] { + return &WaterGeneratorOperator[IN]{ + maxOutOfOrder: maxOutOfOrder, + assigner: assigner, + lastEmittedWatermark: -1, + } +} + +func (w *WaterGeneratorOperator[IN]) Open(ctx context.Context) error { + return nil +} + +func (w *WaterGeneratorOperator[IN]) Process(ctx context.Context, record core.StreamRecord[IN], out Emitter[IN]) error { + if record.Type == core.RecordData { + if err := out.Emit(ctx, record); err != nil { + return err + } + // query the time + eventTime := w.assigner(record.Value) + record.Timestamp = eventTime + + // update the max timestamp + if w.currentMaxTimestamp < eventTime { + w.currentMaxTimestamp = eventTime + } + + // calc the watermark + + watermark := w.currentMaxTimestamp - w.maxOutOfOrder + + // check if current watermark later than previous watermark + if watermark > w.lastEmittedWatermark { + w.lastEmittedWatermark = watermark + return out.Emit(ctx, core.NewWatermarkRecord[IN](watermark)) + } + return nil + } + return out.Emit(ctx, record) +} + +func (w *WaterGeneratorOperator[IN]) Close(ctx context.Context) error { + return nil +} diff --git a/core/operator/window.go b/core/operator/window.go new file mode 100644 index 0000000..610f6e3 --- /dev/null +++ b/core/operator/window.go @@ -0,0 +1,110 @@ +package operator + +import ( + "context" + "fmt" + + "flux/core" +) + +// WindowFunction is for user interface implements the behaviour +type WindowFunction[IN, OUT any] interface { + Apply(ctx context.Context, key string, window core.Window, in []IN, out Emitter[OUT]) error +} + +type WindowOperator[IN, OUT any] struct { + assigner core.WindowAssigner[IN] + userfunc WindowFunction[IN, OUT] + backend core.StateBackend + timeSerivce core.InternalTimerService + + windowState core.ValueState[[]IN] +} + +func NewWindowOperator[IN, OUT any](assigner core.WindowAssigner[IN], userfunc WindowFunction[IN, OUT]) *WindowOperator[IN, OUT] { + return &WindowOperator[IN, OUT]{ + assigner: assigner, + userfunc: userfunc, + timeSerivce: core.NewTimerService(), + } +} + +func (o *WindowOperator[IN, OUT]) Open(ctx context.Context) error { + be, ok := core.ExtractStateBackend(ctx) + if !ok { + return fmt.Errorf("cannot extract state backend, got: %v", be) + } + o.backend = be + windowState, err := be.GetValueState(ctx, "window-elements") + if err != nil { + return fmt.Errorf("cannot get value state window-elements error: %w", err) + } + + o.windowState = core.CastValueType[[]IN](windowState) + + return nil +} + +func (o *WindowOperator[IN, OUT]) Process(ctx context.Context, record core.StreamRecord[IN], out Emitter[OUT]) error { + switch record.Type { + case core.RecordData: + // current windows state will hold all record in window + // assign and get list window + windows := o.assigner.AssignWindows(record.Value, record.Timestamp) + for _, w := range windows { + o.backend.SetCurrentNamespace(w) + o.backend.SetCurrentKey(record.Key) + o.timeSerivce.Register(w.MaxTimestamp(), record.Key, w) + + currentWindowState, err := o.windowState.Value(ctx) + if err != nil { + return err + } + + currentWindowState = append(currentWindowState, record.Value) + + o.windowState.Update(ctx, currentWindowState) + } + + return nil + case core.RecordWatermark: + deadlines := o.timeSerivce.AdvanceWatermark(record.Timestamp) + + for _, timer := range deadlines { + o.backend.SetCurrentKey(timer.Key) + o.backend.SetCurrentNamespace(timer.Namespace) + + currentWindowState, err := o.windowState.Value(ctx) + if err != nil { + return err + } + + // NOTE: this is workaround create inline window, refactor later + + if err := o.userfunc.Apply(ctx, timer.Key, timer.Namespace.(core.Window), currentWindowState, out); err != nil { + return err + } + + if err := o.windowState.Clear(ctx); err != nil { + return err + } + + } + + return out.Emit(ctx, core.StreamRecord[OUT]{ + Timestamp: record.Timestamp, + Type: core.RecordWatermark, + }) + default: + return out.Emit(ctx, core.StreamRecord[OUT]{ + BarrierID: record.BarrierID, + Timestamp: record.Timestamp, + Key: record.Key, + Type: record.Type, + }) + } +} + +func (o *WindowOperator[IN, OUT]) Close(ctx context.Context) error { + return nil +} diff --git a/core/record.go b/core/record.go new file mode 100644 index 0000000..e18e58d --- /dev/null +++ b/core/record.go @@ -0,0 +1,40 @@ +package core + +type RecordType byte + +const ( + RecordData RecordType = iota + RecordWatermark + RecordBarrier +) + +type StreamRecord[T any] struct { + Value T + Key string // for keyBy stateful processing + Timestamp int64 // epoch ms time + BarrierID uint64 // ID checkpoint barrier + Type RecordType +} + +func NewDataRecord[T any](key string, val T, timestamp int64) StreamRecord[T] { + return StreamRecord[T]{ + Value: val, + Key: key, + Timestamp: timestamp, + Type: RecordData, + } +} + +func NewWatermarkRecord[T any](timestamp int64) StreamRecord[T] { + return StreamRecord[T]{ + Type: RecordWatermark, + Timestamp: timestamp, + } +} + +func NewBarrierRecord[T any](barrierID uint64) StreamRecord[T] { + return StreamRecord[T]{ + Type: RecordBarrier, + BarrierID: barrierID, + } +} diff --git a/core/state.go b/core/state.go new file mode 100644 index 0000000..d6d7847 --- /dev/null +++ b/core/state.go @@ -0,0 +1,10 @@ +package core + +import "context" + +// ValueState is for client side using to update state +type ValueState[T any] interface { + Value(ctx context.Context) (T, error) + Update(ctx context.Context, newVal T) (T, error) + Clear(ctx context.Context) error +} diff --git a/core/state_backend.go b/core/state_backend.go new file mode 100644 index 0000000..9f9f397 --- /dev/null +++ b/core/state_backend.go @@ -0,0 +1,62 @@ +package core + +import ( + "context" +) + +type StateBackend interface { + // For Framework === + // SetCurrentKey will perform in Memory + SetCurrentKey(key string) error + // GetCurrentKey will perform in Memory + GetCurrentKey() string + SetCurrentNamespace(namespace any) error + GetCurrentNamespace() any + + // For User === + // GetValueState will perform in disk + GetValueState(ctx context.Context, key string) (ValueState[any], error) +} + +type stateBackendKeyType struct{} + +var stateBackendKey stateBackendKeyType + +type typedValueState[T any] struct { + inner ValueState[any] +} + +func CastValueType[T any](inner ValueState[any]) ValueState[T] { + return &typedValueState[T]{inner: inner} +} + +func (s *typedValueState[T]) Value(ctx context.Context) (T, error) { + val, err := s.inner.Value(ctx) + if err != nil { + var zero T + return zero, err + } + if val == nil { + var zero T + return zero, nil + } + return val.(T), nil +} + +func (s *typedValueState[T]) Update(ctx context.Context, newVal T) (T, error) { + _, err := s.inner.Update(ctx, newVal) + return newVal, err +} + +func (s *typedValueState[T]) Clear(ctx context.Context) error { + return s.inner.Clear(ctx) +} + +func InjectStateBackend(ctx context.Context, state StateBackend) context.Context { + return context.WithValue(ctx, stateBackendKey, state) +} + +func ExtractStateBackend(ctx context.Context) (StateBackend, bool) { + backend, ok := ctx.Value(stateBackendKey).(StateBackend) + return backend, ok +} diff --git a/core/timer.go b/core/timer.go new file mode 100644 index 0000000..3d17299 --- /dev/null +++ b/core/timer.go @@ -0,0 +1,113 @@ +package core + +import ( + "container/heap" + "context" +) + +// If using rockDB no need priority queue anymore +// implement priority queue from golang +// ref: https://pkg.go.dev/container/heap#example-package-PriorityQueue + +type timerServiceKey struct{} + +// InjectTimerService inject the timer into Context +func InjectTimerService(ctx context.Context, ts TimerService) context.Context { + return context.WithValue(ctx, timerServiceKey{}, ts) +} + +// ExtractTimerService get the timer for user from context +func ExtractTimerService(ctx context.Context) (TimerService, bool) { + ts, ok := ctx.Value(timerServiceKey{}).(TimerService) + return ts, ok +} + +type Timer struct { + // Timestamp is the time when Timer is triggered + Timestamp int64 + // Key is for who User involve + Key string + // Window is for time range start-end of + // String actually cost more bytes than Window(16bytes) + // and we dont want to decode from namespace to Window + Namespace any +} + +// Priority Queue (Min Heap). Smallest timestamp will be on top + +type timerHeap []Timer + +func (h *timerHeap) Len() int { return len(*h) } +func (h *timerHeap) Less(i, j int) bool { return (*h)[i].Timestamp < (*h)[j].Timestamp } +func (h *timerHeap) Swap(i, j int) { + (*h)[i], (*h)[j] = (*h)[j], (*h)[i] +} + +func (h *timerHeap) Push(x any) { + *h = append(*h, x.(Timer)) +} + +func (h *timerHeap) Pop() any { + old := *h + n := len(old) + last := old[n-1] + *h = old[:n-1] + return last +} + +type TimerService interface { + Register(time int64, key string, keyspace any) + CurrentWatermark() int64 +} + +type InternalTimerService interface { + TimerService + AdvanceWatermark(watermark int64) []Timer +} + +type timerService struct { + queue timerHeap + currentWatermark int64 +} + +func NewTimerService() InternalTimerService { + return &timerService{ + queue: make(timerHeap, 0), + currentWatermark: -1, + } +} + +func (t *timerService) Register(time int64, key string, namespace any) { + heap.Push(&t.queue, Timer{ + Timestamp: time, + Key: key, + Namespace: namespace, + }) +} + +func (t *timerService) CurrentWatermark() int64 { + return t.currentWatermark +} + +func (t *timerService) AdvanceWatermark(watermark int64) []Timer { + if watermark < t.currentWatermark { + // no need to advance when get smaller + return nil + } + + t.currentWatermark = watermark + + var timers []Timer + for t.queue.Len() > 0 { + earliestTimer := t.queue[0] + + if earliestTimer.Timestamp > watermark { + break + } + + pop := heap.Pop(&t.queue).(Timer) + timers = append(timers, pop) + } + + return timers +} diff --git a/core/window.go b/core/window.go new file mode 100644 index 0000000..6c4b98f --- /dev/null +++ b/core/window.go @@ -0,0 +1,54 @@ +package core + +import "fmt" + +type Window interface { + // MaxTimestamp is the maximum time the Window is Open + MaxTimestamp() int64 + // ToString convert Window to string format + ToString() string +} + +type TimeWindow struct { + Start int64 + End int64 +} + +func (t TimeWindow) MaxTimestamp() int64 { + // MaxTimestamp is 1ms before the End + return t.End - 1 +} + +func (t TimeWindow) ToString() string { + return fmt.Sprintf("timewindow(%d,%d)", t.Start, t.End) +} + +type WindowAssigner[IN any] interface { + AssignWindows(element IN, timestamp int64) []Window +} + +type TumblingWindows[IN any] struct { + Size int64 +} + +func NewTumblingWindows[IN any](size int64) *TumblingWindows[IN] { + return &TumblingWindows[IN]{ + Size: size, + } +} + +func (w *TumblingWindows[IN]) AssignWindows(element IN, timestamp int64) []Window { + // module to get start end + // if timestamp = 1200, size 1000 + // so start will 1200 - 200 = 1000, end will 1000 + 1000 = 2000 + start := timestamp - (timestamp % w.Size) + + end := start + w.Size + + return []Window{ + TimeWindow{ + Start: start, + End: end, + }, + } +} -- cgit v1.3.1