StreamSQL Custom Functions

StreamSQL provides a plugin-style custom function system: register functions at runtime from Go — no restart needed — and use them in SQL immediately.

Core Features

• Unified registry: aggregation, analytical, and scalar functions all register the same way
• Runtime registration: functions.Register / functions.RegisterCustomFunction, no restart
• Type safety: argument-count validation and type-conversion helpers
• Stateful analytics: cross-event state via the StatefulAnalytic interface

Function Types

Type	Constant	Use	Example
Aggregation	TypeAggregation	Windowed aggregation	sum(), avg(), count()
Analytical	TypeAnalytical	Cross-event stateful analysis	lag(), had_changed()
Window	TypeWindow	Window metadata	window_start(), window_end()
Math	TypeMath	Numeric calculation	abs(), round()
String	TypeString	Text processing	upper(), concat()
Conversion	TypeConversion	Type conversion	cast(), to_json()
Datetime	TypeDateTime	Time handling	now(), date_format()
Custom (scalar)	TypeCustom	General scalar logic	business-specific

1. Custom Aggregate Function

Aggregate functions implement the AggregatorFunction interface (New / Add / Result / Reset / Clone) and are used inside windowed / GROUP BY queries.

package main

import (
    "github.com/rulego/streamsql/functions"
    "github.com/rulego/streamsql/utils/cast"
)

// CustomProduct — product of values
type CustomProduct struct {
    *functions.BaseFunction
    product float64
    first   bool
}

func NewCustomProduct() *CustomProduct {
    return &CustomProduct{
        BaseFunction: functions.NewBaseFunction("product", functions.TypeAggregation,
            "custom aggregate", "product of values", 1, -1),
        product: 1.0,
        first:   true,
    }
}

func (f *CustomProduct) Validate(args []any) error { return f.ValidateArgCount(args) }
func (f *CustomProduct) Execute(ctx *functions.FunctionContext, args []any) (any, error) {
    return f.Result(), nil
}

// AggregatorFunction interface
func (f *CustomProduct) New() functions.AggregatorFunction {
    return &CustomProduct{BaseFunction: f.BaseFunction, product: 1.0, first: true}
}
func (f *CustomProduct) Add(value any) {
    if v, err := cast.ToFloat64E(value); err == nil {
        if f.first {
            f.product = v
            f.first = false
        } else {
            f.product *= v
        }
    }
}
func (f *CustomProduct) Result() any {
    if f.first {
        return 0.0
    }
    return f.product
}
func (f *CustomProduct) Reset() { f.product = 1.0; f.first = true }
func (f *CustomProduct) Clone() functions.AggregatorFunction {
    return &CustomProduct{BaseFunction: f.BaseFunction, product: f.product, first: f.first}
}

func main() {
    functions.Register(NewCustomProduct())
    // SELECT device, product(value) AS p FROM stream GROUP BY device, TumblingWindow('1m')
}

2. Custom Scalar Function (simple closure)

For stateless per-row computation, use RegisterCustomFunction with a closure — no struct needed.

functions.RegisterCustomFunction("double", functions.TypeMath,
    "math", "double the value", 1, 1,
    func(ctx *functions.FunctionContext, args []any) (any, error) {
        v, err := cast.ToFloat64E(args[0])
        if err != nil {
            return nil, err
        }
        return v * 2, nil
    })

// SELECT double(temperature) AS d FROM stream

3. Custom Analytical Function (StatefulAnalytic)

Analytical functions need cross-event state ("previous value", "running sum", "has it changed"); they cannot be stateless closures. Implement StatefulAnalytic: NewState returns a state object, the engine holds one per partition and calls Apply on every event. Register with functions.Register (no RegisterAnalyticalAdapter needed).

package main

import (
    "fmt"

    "github.com/rulego/streamsql/functions"
    "github.com/rulego/streamsql/utils/cast"
)

// MovingAverage — moving average over the last N values
type MovingAverage struct {
    *functions.BaseFunction
    windowSize int
}

func NewMovingAverage(windowSize int) *MovingAverage {
    return &MovingAverage{
        BaseFunction: functions.NewBaseFunction("moving_avg", functions.TypeAnalytical,
            "custom analytic", "moving average", 1, 1),
        windowSize: windowSize,
    }
}

func (f *MovingAverage) Validate(args []any) error { return f.ValidateArgCount(args) }

// Execute is disabled on the scalar path: analytics need cross-row state.
func (f *MovingAverage) Execute(ctx *functions.FunctionContext, args []any) (any, error) {
    return nil, fmt.Errorf("analytic function %q must be used as a field or with OVER", f.GetName())
}

// NewState implements StatefulAnalytic: one independent state per partition.
func (f *MovingAverage) NewState() functions.AnalyticState {
    return &movingAvgState{windowSize: f.windowSize}
}

type movingAvgState struct {
    windowSize int
    values     []float64
}

func (s *movingAvgState) Apply(args []any) any {
    if len(args) == 0 {
        return nil
    }
    v, err := cast.ToFloat64E(args[0])
    if err != nil {
        return nil
    }
    s.values = append(s.values, v)
    if len(s.values) > s.windowSize {
        s.values = s.values[len(s.values)-s.windowSize:]
    }
    sum := 0.0
    for _, x := range s.values {
        sum += x
    }
    return sum / float64(len(s.values))
}

func (s *movingAvgState) Reset() { s.values = nil }

func main() {
    functions.Register(NewMovingAverage(5))
    // SELECT device, moving_avg(temperature) AS ma FROM stream
    // SELECT moving_avg(temperature) OVER (PARTITION BY device) AS ma FROM stream
}

Scalar vs analytical

If a function only uses the current row's arguments (Z-score, health score) and keeps no state — use TypeMath / TypeCustom + RegisterCustomFunction. Only use TypeAnalytical + StatefulAnalytic when you need cross-event state (previous value, running accumulation, change detection).

Function Management

// Check existence
if _, exists := functions.Get("double"); exists {
    fmt.Println("registered")
}

// Inspect
if fn, exists := functions.Get("double"); exists {
    fmt.Printf("type: %s\n", fn.GetType())
}

// Unregister
functions.Unregister("double")

📚 Related Documentation

• Analytical Functions - Built-in analytic functions and the OVER clause
• Aggregate Functions - Built-in aggregate functions
• SQL Reference - Complete SQL syntax reference