Schema Validation

The github.com/rulego/streamsql/schema package provides a runtime registry for typed record schemas, used to validate map[string]interface{} against declared field definitions before the data enters the stream.

Overview

A stream processing pipeline carries implicit assumptions about upstream field names and types: aggregate functions expect numbers, time windows expect timestamps, and GROUP BY expects keys to be present. When an upstream sends rows with missing fields or wrong types, the error often surfaces only when a window fires or an aggregate runs, which makes it hard to debug.

Schema validation addresses exactly this — before calling ssql.Emit, you run each map through an explicit field declaration and reject or log non-conforming rows so dirty data never enters the stream.

Schema vs. SQL WHERE

WHERE is a row-level filter that operates on data already inside the stream, keeping or dropping rows by condition. Schema validation operates at the edge, before the stream, judging whether the whole row structure is well-formed. They complement each other: Schema guards the door, WHERE selects inside.

Define and Register a Schema

package schema

type FieldDef struct {
    Name     string
    Type     DataType
    Required bool
    Default  interface{} // non-nil suppresses the required-missing error; use a typed value, e.g. float64(0)
}

type Schema struct {
    Name   string      // unique within a Registry
    Fields []FieldDef  // ordered
    Strict bool        // when true, unknown keys are errors
}

Below is an iot schema with a required id (int), an optional note (string), a score (float) with a default, and an arbitrary meta field:

package main

import "github.com/rulego/streamsql/schema"

func main() {
    iot := schema.Schema{
        Name: "iot",
        Fields: []schema.FieldDef{
            {Name: "id",    Type: schema.TypeInt,   Required: true},
            {Name: "note",  Type: schema.TypeString},
            {Name: "score", Type: schema.TypeFloat, Default: float64(0)},
            {Name: "meta",  Type: schema.TypeAny},
        },
    }
    if err := schema.Default.Register(iot); err != nil {
        panic(err)
    }
}

Default and nil

Whether Default takes effect depends on whether it is nil. To use a zero value as the default you must wrap a typed literal into the interface: float64(0), int(0), and false are all non-nil and suppress the required-missing error; a bare nil means "no default" and a required field that is absent still errors. Keep the type consistent too — writing int(0) for a TypeFloat field suppresses the missing error but the value type won't line up.

Validate Input

func (s *Schema) Validate(data map[string]interface{}) error

Validate aggregates every problem into one return instead of stopping at the first error:

A clean row:

err := schema.Default.MustGet("iot").Validate(map[string]interface{}{
    "id":    42,
    "note":  "ok",
    "score": 9.5,
    "meta":  nil,   // TypeAny accepts nil
})
// err == nil

A failing row with several problems at once, aggregated into one error:

err := schema.Default.MustGet("iot").Validate(map[string]interface{}{
    // id is missing (required, no default)
    "note":  123,    // expects string, got int
    "score": "high", // expects float, got string
})
// schema "iot": required field "id" is missing; schema "iot": field "note" expects string, got int; schema "iot": field "score" expects float, got string

When more than one problem is found, Validate returns a *MultiError:

type MultiError struct{ Errors []error }

func (m *MultiError) Append(err error)  // appends only when err is non-nil
func (m *MultiError) Err() error        // returns nil when empty, otherwise m itself
func (m *MultiError) Error() string     // joins sub-errors with "; "

Strict Mode

By default (Strict=false), keys in the input that are not declared are silently ignored, which lets upstream fields evolve gradually. With Strict=true, unknown keys are errors:

strict := schema.Schema{
    Name:   "strict_iot",
    Strict: true,
    Fields: iot.Fields,
}
schema.Default.Register(strict)

err := schema.Default.MustGet("strict_iot").Validate(map[string]interface{}{
    "id":    1,
    "extra": "not declared",
})
// schema "strict_iot": unknown field "extra"

Registry Usage

type Registry struct{ ... }  // guarded by sync.RWMutex, concurrent-safe

func NewRegistry() *Registry
func (r *Registry) Register(s Schema) error      // errors on empty name or duplicate
func (r *Registry) Get(name string) (Schema, bool)
func (r *Registry) MustGet(name string) Schema   // panics when missing

var Default = NewRegistry()  // package-level shared registry

• Default: use directly in most cases; schemas are shared across components.
• NewRegistry(): for tests or multi-tenant scenarios where you need isolated schemas.
• MustGet: only when the invariant "this schema is definitely registered" holds (e.g. registered in init); otherwise prefer Get and handle the false branch.

Register never overwrites

Register returns an error for a duplicate name and does not silently overwrite. To update a definition, pick a new name or create a fresh Registry.

Complete Example

After registering a schema, validate on every Emit to drop or log invalid rows before they enter the stream:

package main

import (
    "fmt"

    "github.com/rulego/streamsql"
    "github.com/rulego/streamsql/schema"
)

func main() {
    s := schema.Schema{
        Name: "iot",
        Fields: []schema.FieldDef{
            {Name: "deviceId",    Type: schema.TypeString, Required: true},
            {Name: "temperature", Type: schema.TypeFloat,  Required: true},
            {Name: "humidity",    Type: schema.TypeFloat,  Default: float64(0)},
            {Name: "tags",        Type: schema.TypeArray},
        },
    }
    if err := schema.Default.Register(s); err != nil {
        panic(err)
    }

    ssql := streamsql.New()
    defer ssql.Stop()
    ssql.Execute(`SELECT deviceId, AVG(temperature) AS avg_t
                  FROM stream
                  GROUP BY deviceId, TumblingWindow('5s')`)
    ssql.AddSink(func(rows []map[string]interface{}) { fmt.Println(rows) })

    row := map[string]interface{}{
        "deviceId":    "d1",
        "temperature": 25.5,
    }
    if err := schema.Default.Get("iot").Validate(row); err != nil {
        fmt.Println("invalid:", err)
        return
    }
    ssql.Emit(row)
}

DataType Reference

type DataType int

const (
    TypeInt DataType = iota
    TypeInt64
    TypeFloat
    TypeBool
    TypeString
    TypeTime
    TypeArray
    TypeMap
    TypeAny
)

Type	String form	Matching Go value	Notes
TypeInt	int	int	The three numeric types are interchangeable: a TypeInt/TypeInt64/TypeFloat field accepts any numeric value
TypeInt64	int64	int64	see above
TypeFloat	float	float32, float64	see above
TypeBool	bool	bool
TypeString	string	string
TypeTime	time	time.Time
TypeArray	array	[]interface{}	Only []interface{} is recognized; other slice types fall back to TypeAny
TypeMap	map	map[string]interface{}	Only map[string]interface{} is recognized
TypeAny	any	any value (including nil)	Catch-all; InferType also returns TypeAny for nil and unrecognized values

func InferType(v interface{}) DataType

InferType maps a runtime value to a DataType: nil and unrecognized values return TypeAny, both float32 and float64 map to TypeFloat, and only []interface{} and map[string]interface{} are recognized as containers.

Numeric interchangeability

A field declared TypeInt accepts any of int/int64/float32/float64, and the same goes for TypeInt64 and TypeFloat. This means a float64 decoded from upstream JSON is not flagged as a type mismatch.

📚 Related Documentation

• API Reference — core StreamSQL API
• SQL Reference — full SQL syntax