自定义函数

StreamSQL 提供插件式自定义函数系统：在 Go 中运行时注册函数，无需重启，注册后立即可在 SQL 中使用。

核心特性

• 统一注册：聚合、分析、标量函数注册方式一致
• 运行时注册：functions.Register / functions.RegisterCustomFunction，无需重启
• 类型安全：参数个数校验与类型转换辅助
• 有状态分析：通过 StatefulAnalytic 接口实现跨事件状态

函数类型

类型	常量	用途	示例
聚合函数	TypeAggregation	窗口聚合	sum(), avg(), count()
分析函数	TypeAnalytical	跨事件状态分析	lag(), had_changed()
窗口函数	TypeWindow	窗口元数据	window_start(), window_end()
数学函数	TypeMath	数值计算	abs(), round()
字符串函数	TypeString	文本处理	upper(), concat()
转换函数	TypeConversion	类型转换	cast(), to_json()
时间函数	TypeDateTime	时间处理	now(), date_format()
通用（标量）函数	TypeCustom	通用标量逻辑	业务自定义

注册入口

无论哪种类型，都只需 functions.Register 一个入口：标量函数也可用 functions.RegisterCustomFunction 传闭包。聚合函数实现 AggregatorFunction 后 Register，适配器会自动接通，无需 aggregator.Register 或 RegisterAggregatorAdapter；分析函数实现 StatefulAnalytic 后 Register，无需 RegisterAnalyticalAdapter。

1. 自定义聚合函数

聚合函数实现 AggregatorFunction 接口（New / Add / Result / Reset / Clone），在窗口 / GROUP BY 查询中使用。

package main

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

// CustomProduct 计算数值乘积
type CustomProduct struct {
    *functions.BaseFunction
    product float64
    first   bool
}

func NewCustomProduct() *CustomProduct {
    return &CustomProduct{
        BaseFunction: functions.NewBaseFunction("product", functions.TypeAggregation,
            "自定义聚合函数", "计算数值乘积", 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 接口
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. 自定义标量函数（闭包）

无状态的逐行计算，用 RegisterCustomFunction 注册闭包，无需定义结构体。

functions.RegisterCustomFunction("double", functions.TypeMath,
    "数学函数", "将值乘以2", 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. 自定义分析函数（StatefulAnalytic）

分析函数需要跨事件状态（"前一个值"、"累积求和"、"是否变化"），不能用无状态闭包。实现 StatefulAnalytic：NewState 返回一份状态，引擎为每个分区各持一份、逐条事件调 Apply。用 functions.Register 注册（无需 RegisterAnalyticalAdapter）。

package main

import (
    "fmt"

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

// MovingAverage 维护最近 N 个值的窗口
type MovingAverage struct {
    *functions.BaseFunction
    windowSize int
}

func NewMovingAverage(windowSize int) *MovingAverage {
    return &MovingAverage{
        BaseFunction: functions.NewBaseFunction("moving_avg", functions.TypeAnalytical,
            "自定义分析函数", "窗口移动平均", 1, 1),
        windowSize: windowSize,
    }
}

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

// Execute 标量路径禁用：分析函数需跨行状态，由状态机求值，不能嵌在标量表达式里。
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 实现 StatefulAnalytic：每个分区一份独立窗口状态。
func (f *MovingAverage) NewState() functions.AnalyticState {
    return &movingAvgState{windowSize: f.windowSize}
}

type movingAvgState struct {
    windowSize int
    values     []float64
}

// Apply 每条事件调用：加入新值、保持窗口大小、返回当前移动平均。
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
}

标量 vs 分析

如果函数只基于当前行参数算一个结果（如 Z-Score、健康分），不跨事件——用 TypeMath / TypeCustom + RegisterCustomFunction（标量闭包）。只有需要跨事件状态（前值 / 累积 / 变化）才用 TypeAnalytical + StatefulAnalytic。

函数管理

// 检查是否存在
if _, exists := functions.Get("double"); exists {
    fmt.Println("已注册")
}

// 查看函数信息
if fn, exists := functions.Get("double"); exists {
    fmt.Printf("类型: %s\n", fn.GetType())
    fmt.Printf("描述: %s\n", fn.GetDescription())
}

// 注销
functions.Unregister("double")

// 列出所有已注册函数
allFunctions := functions.ListAll()
for name, fn := range allFunctions {
    fmt.Printf("函数: %s, 类型: %s\n", name, fn.GetType())
}

📚 相关文档

• 分析函数 - 内置分析函数与 OVER 子句
• 聚合函数 - 内置聚合函数
• SQL 参考 - 完整 SQL 语法参考
• 案例集锦 - 端到端应用案例