Schema 校验

github.com/rulego/streamsql/schema 包提供一个运行期类型化记录 schema 注册表，用于在数据进入流之前对 map[string]interface{} 做校验，把不合规的输入挡在边缘。

概述

流处理管道对上游数据的字段名和类型通常有隐式假设：聚合函数期望数值，时间窗口期望时间戳，GROUP BY 期望键存在。一旦上游送来缺字段或类型错乱的行，错误往往要等到窗口触发或聚合计算时才暴露，排查困难。

Schema 校验解决的就是这个问题——在调用 ssql.Emit 之前，先用一份显式声明的字段定义把每行 map 过一遍，不合规就拒绝或记录，让脏数据进不了流。

与 SQL WHERE 的区别

WHERE 是行级过滤器，作用在已经进入流的数据上，按条件保留或丢弃行；Schema 校验作用在流的入口之前，按字段定义判断整行结构是否合法。两者互补：Schema 守门，WHERE 筛选。

定义并注册 Schema

package schema

type FieldDef struct {
    Name     string
    Type     DataType
    Required bool
    Default  interface{} // 非 nil 时屏蔽 required 缺失错误；用带类型的值，如 float64(0)
}

type Schema struct {
    Name   string      // 在同一个 Registry 内唯一
    Fields []FieldDef  // 有序
    Strict bool        // 为 true 时，未知键视为错误
}

下面声明一个名为 iot 的 schema：必填的 id（int）、可选的 note（string）、带默认值的 score（float）以及一个任意类型的 meta 字段。

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 与 nil

Default 是否生效取决于它是否为 nil。要让零值作为默认值，必须用带类型的字面量装进 interface：float64(0)、int(0)、false 都是非 nil，能屏蔽 required 缺失错误；而裸的 nil 表示"没有默认值"，必填字段缺键仍会报错。注意类型也要对上——给 TypeFloat 字段写 int(0) 虽然能屏蔽缺失错误，但取值时类型不匹配。

校验输入

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

Validate 一次性聚合所有问题返回，不会遇到第一个错误就停：

通过的一行：

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

不通过的一行（同时多处错误，聚合成一条返回）：

err := schema.Default.MustGet("iot").Validate(map[string]interface{}{
    // id 缺失（必填，无默认）
    "note":  123,    // 期望 string，得到 int
    "score": "high", // 期望 float，得到 string
})
// schema "iot": required field "id" is missing; schema "iot": field "note" expects string, got int; schema "iot": field "score" expects float, got string

Validate 在发现多于一个问题时返回 *MultiError：

type MultiError struct{ Errors []error }

func (m *MultiError) Append(err error)       // 非 nil 才追加
func (m *MultiError) Err() error             // 无错误返回 nil，否则返回 m 自身
func (m *MultiError) Error() string          // 用 "; " 连接所有子错误

Strict 模式

默认（Strict=false）下，输入里未声明的键被静默忽略，方便上游逐步演进字段。开启 Strict=true 后，未知键即报错：

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 用法

type Registry struct{ ... }  // sync.RWMutex，并发安全

func NewRegistry() *Registry
func (r *Registry) Register(s Schema) error     // 名字为空或重复时返回错误
func (r *Registry) Get(name string) (Schema, bool)
func (r *Registry) MustGet(name string) Schema  // 找不到时 panic

var Default = NewRegistry()  // 包级共享注册表

• Default：多数场景直接用，跨组件共享 schema。
• NewRegistry()：需要在测试或多租户场景隔离 schema 时，用私有注册表。
• MustGet：仅在"schema 一定已注册"这一前置不变量成立时用（比如 init 阶段注册）；否则优先 Get 处理 false 分支。

Register 不可覆盖

Register 对同名 schema 返回错误，不会静默覆盖。需要更新定义时，先换名字或新建一个 Registry。

完整示例

注册 schema 后，在每次 Emit 之前先校验，把不合规的行挡掉/记日志：

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 参考

type DataType int

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

类型	字符串形式	匹配的 Go 值	说明
TypeInt	int	int	三种数值类型互通：TypeInt/TypeInt64/TypeFloat 字段互相接受任意数值
TypeInt64	int64	int64	见上
TypeFloat	float	float32、float64	见上
TypeBool	bool	bool
TypeString	string	string
TypeTime	time	time.Time
TypeArray	array	[]interface{}	仅识别 []interface{}；其他切片类型回退为 TypeAny
TypeMap	map	map[string]interface{}	仅识别 map[string]interface{}
TypeAny	any	任意值（含 nil）	兜底类型；InferType 对 nil 和未识别值也返回 TypeAny

func InferType(v interface{}) DataType

InferType 把运行时值映射到 DataType：nil 和未识别的值返回 TypeAny，float32/float64 都映射到 TypeFloat，只识别 []interface{} 和 map[string]interface{} 两种容器。

数值类型互通

声明为 TypeInt 的字段接受 int/int64/float32/float64 中的任意一种，TypeInt64 和 TypeFloat 同理。这样上游 JSON 解析出的 float64 不会被误判为类型不匹配。

📚 相关文档

• API 参考 — StreamSQL 核心 API
• SQL 参考 — 完整 SQL 语法