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/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 +++++++++++++++++++++++++++++++++++++++++ 8 files changed, 464 insertions(+) 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 (limited to 'core/operator') 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 +} -- cgit v1.3.1