From 91b1c642601efbbcc5af9931e087f68b2c72fc3d Mon Sep 17 00:00:00 2001 From: SangTran-127 Date: Thu, 13 Aug 2026 23:42:53 +0700 Subject: feat: add on timer & watermark --- core/filter.go | 33 ------------- core/operator/filter.go | 36 ++++++++++++++ core/operator/keyed_process.go | 81 +++++++++++++++++++++++++++++++ core/operator/watermark.go | 58 ++++++++++++++++++++++ core/rich_map.go | 1 - core/state_backend.go | 34 +++++++++++++ core/timer.go | 108 +++++++++++++++++++++++++++++++++++++++++ 7 files changed, 317 insertions(+), 34 deletions(-) delete mode 100644 core/filter.go create mode 100644 core/operator/filter.go create mode 100644 core/operator/keyed_process.go create mode 100644 core/operator/watermark.go delete mode 100644 core/rich_map.go create mode 100644 core/timer.go (limited to 'core') diff --git a/core/filter.go b/core/filter.go deleted file mode 100644 index 6c95fbc..0000000 --- a/core/filter.go +++ /dev/null @@ -1,33 +0,0 @@ -package core - -import "context" - -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 StreamRecord[IN], out Emitter[IN]) error { - if record.Type != 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/filter.go b/core/operator/filter.go new file mode 100644 index 0000000..183148b --- /dev/null +++ b/core/operator/filter.go @@ -0,0 +1,36 @@ +package operator + +import ( + "context" + "goflink/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/keyed_process.go b/core/operator/keyed_process.go new file mode 100644 index 0000000..b507d87 --- /dev/null +++ b/core/operator/keyed_process.go @@ -0,0 +1,81 @@ +package operator + +import ( + "context" + "fmt" + "goflink/core" +) + +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/watermark.go b/core/operator/watermark.go new file mode 100644 index 0000000..75390a8 --- /dev/null +++ b/core/operator/watermark.go @@ -0,0 +1,58 @@ +package operator + +import ( + "context" + "goflink/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 { + + // 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/rich_map.go b/core/rich_map.go deleted file mode 100644 index 9a8bc95..0000000 --- a/core/rich_map.go +++ /dev/null @@ -1 +0,0 @@ -package core diff --git a/core/state_backend.go b/core/state_backend.go index f8b83fb..729f0fe 100644 --- a/core/state_backend.go +++ b/core/state_backend.go @@ -3,11 +3,15 @@ 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 string) error + GetCurrentNamespace() string + // For User === // GetValueState will perform in disk GetValueState(ctx context.Context, key string) (ValueState[any], error) } @@ -16,6 +20,36 @@ 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) } diff --git a/core/timer.go b/core/timer.go new file mode 100644 index 0000000..f744382 --- /dev/null +++ b/core/timer.go @@ -0,0 +1,108 @@ +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 +} + +// 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) + 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) { + heap.Push(&t.queue, Timer{ + Timestamp: time, + Key: key, + }) +} + +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 +} -- cgit v1.2.3