Advanced Examples

This page collects advanced usage such as custom functions and performance modes. For basic onboarding, see Quick Start.

Custom Functions

StreamSQL supports registering custom functions (UDFs); once registered they can be used in SQL just like built-in functions.

package main

import (
    "fmt"
    "math"
    "time"

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

func main() {
    registerCustomFunctions() // register the UDF first

    ssql := streamsql.New()
    defer ssql.Stop()

    sql := `SELECT
                device,
                square(value)         AS squared_value,
                f_to_c(temperature)   AS celsius,
                circle_area(radius)   AS area
            FROM stream
            WHERE value > 0`
    if err := ssql.Execute(sql); err != nil {
        panic(err)
    }

    ssql.AddSink(func(results []map[string]interface{}) {
        fmt.Printf("Custom function results: %v\n", results)
    })

    testData := []map[string]interface{}{
        {"device": "sensor1", "value": 5.0, "temperature": 68.0, "radius": 3.0},
        {"device": "sensor2", "value": 10.0, "temperature": 86.0, "radius": 2.5},
    }
    for _, data := range testData {
        ssql.Emit(data)
        time.Sleep(200 * time.Millisecond)
    }
    time.Sleep(500 * time.Millisecond)
}

// Register custom functions
func registerCustomFunctions() {
    // Math function: square
    functions.RegisterCustomFunction(
        "square", functions.TypeMath, "Math Function", "Calculate square", 1, 1,
        func(ctx *functions.FunctionContext, args []interface{}) (interface{}, error) {
            val := cast.ToFloat64(args[0])
            return val * val, nil
        },
    )
    // Fahrenheit to Celsius
    functions.RegisterCustomFunction(
        "f_to_c", functions.TypeConversion, "Temperature Conversion", "Fahrenheit to Celsius", 1, 1,
        func(ctx *functions.FunctionContext, args []interface{}) (interface{}, error) {
            fahrenheit := cast.ToFloat64(args[0])
            return (fahrenheit - 32) * 5 / 9, nil
        },
    )
    // Circle area
    functions.RegisterCustomFunction(
        "circle_area", functions.TypeMath, "Geometric Calculation", "Calculate circle area", 1, 1,
        func(ctx *functions.FunctionContext, args []interface{}) (interface{}, error) {
            radius := cast.ToFloat64(args[0])
            if radius < 0 {
                return nil, fmt.Errorf("radius must be positive")
            }
            return math.Pi * radius * radius, nil
        },
    )
}

Key points:

• functions.RegisterCustomFunction(name, type, category, desc, minArgs, maxArgs, fn) registers a UDF; it returns (any, error).
• Function types: TypeMath / TypeConversion / TypeString / TypeDateTime / TypeAggregation / TypeAnalytical / TypeCustom.
• After registration, call them in SELECT/WHERE like built-in functions (e.g. square(value), f_to_c(temperature)).
• Use cast.ToFloat64 and friends for tolerant argument conversion; registration must complete before Execute.

Performance Modes

StreamSQL provides several preset performance modes; pick one per scenario:

package main

import (
    "fmt"

    "github.com/rulego/streamsql"
)

func main() {
    // High performance mode: large buffers + expand strategy, suited for high throughput
    ssqlHighPerf := streamsql.New(streamsql.WithHighPerformance())
    defer ssqlHighPerf.Stop()

    // Low latency mode: small buffers + fast response, suited for real-time alerts
    ssqlLowLatency := streamsql.New(streamsql.WithLowLatency())
    defer ssqlLowLatency.Stop()

    sql := "SELECT deviceId, AVG(temperature) FROM stream GROUP BY deviceId, TumblingWindow('5s')"
    ssqlHighPerf.Execute(sql)
    ssqlLowLatency.Execute(sql)

    fmt.Println("Different performance modes started")
}

Mode	Suited Scenario	Characteristics
Default New()	General	Balanced buffer and worker configuration
WithHighPerformance()	High throughput	Large buffers, expand overflow strategy
WithLowLatency()	Real-time response	Small buffers, block overflow, low latency

For finer control, use streamsql.NewStreamWithCustomPerformance(config) to customize buffer size, worker count, overflow strategy (drop/block/expand), etc.

📚 Related Docs

• Quick Start
• API Reference — performance config options, RegisterCustomFunction