streamsql

package module
v1.1.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 14, 2026 License: Apache-2.0 Imports: 10 Imported by: 1

README

StreamSQL

GoDoc Go Report CI RELEASE codecov Mentioned in Awesome Go

English | 简体中文

StreamSQL is an embeddable, SQL-based stream processing engine built for the IoT edge. It sits between a time-series database and Apache Flink: Flink-grade real-time computation with TSDB-grade lightweight deployment — run real-time filtering, windowed aggregation, CDC-style change detection, and complex event pattern matching, in-process, inside a 128MB gateway.

The real-time power of a TSDB + the computation power of Flink + minimal deployment and integration overhead.

📖 Documentation | Similar to: Apache Flink

  • 🪶 Lightweight & embeddable — pure in-memory, zero external deps, fits a 128MB gateway as a library, starts in seconds
  • 🧩 Full SQL — tumbling/sliding/counting/session/global windows, event time + watermark, CASE, nested fields, HAVING
  • 🔍 Analytic functions — lag / had_changed / changed_col / cumulative stats, purpose-built for CDC change detection and context backtracking
  • 🧩 Complex Event Processing (CEP) — MATCH_RECOGNIZE (SQL:2016, Flink-aligned) — unique among lightweight edge engines
  • 🚀 Edge-grade performance — ~1.92M msg/s single-core filtering on x86; 128MB holds 100k+ device partitions
  • 🔌 RuleGo ecosystem — tap RuleGo components for MQTT / HTTP / message queues / databases and any data source

Why StreamSQL

Traditional stream processing forces two extremes: time-series databases store well but compute weakly in real time; Flink / Storm are powerful but heavy, consuming GBs of memory — unsuitable for the edge. StreamSQL fills the gap — purpose-built for the edge, doing real-time aggregation and pattern recognition on massive data under tight resource constraints.

StreamSQL Apache Flink eKuiper Time-series DB
Deployment Minimal Complex Simple Moderate
Footprint Tiny (~10MB) High (GBs) Tiny (~10MB) Moderate
Embeddable / as a library ✅ ❌ ⚠️ ⚠️
Full SQL ✅ ✅ ✅ Limited
Complex Event Processing (CEP) ✅ ✅ ❌ ❌
Analytic / change detection ✅ ✅ ✅ ❌
Event time + watermark ✅ ✅ ⚠️ ❌
Edge deployment ✅ ❌ ✅ ⚠️
Horizontal cluster scaling Single-node ✅ Single-node ✅

Good fit: edge real-time compute on IoT gateways / industrial controllers / vehicle systems, device monitoring & anomaly detection, stream-processing prototyping, adding SQL muscle to RuleGo rule chains. Not a fit: large-scale clusters needing horizontal scaling, apps needing persisted state or ACID transactions.

Installation

go get github.com/rulego/streamsql

Quick Start

Each record is processed and emitted immediately — real-time transformation and filtering, no window wait:

package main

import (
	"fmt"
	"github.com/rulego/streamsql"
)

func main() {
	ssql := streamsql.New()
	defer ssql.Stop()

	err := ssql.Execute(`SELECT deviceId,
	    temperature * 1.8 + 32 AS fahrenheit,
	    CASE WHEN temperature > 30 THEN 'hot' ELSE 'normal' END AS level
	    FROM stream WHERE temperature > 0`)
	if err != nil {
		panic(err)
	}

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

	ssql.Emit(map[string]interface{}{"deviceId": "sensor01", "temperature": 32.5})
}
// => Result: map[deviceId:sensor01 fahrenheit:90.5 level:hot]

Core Capabilities

🧩 Complex Event Processing (CEP) — unique among lightweight edge engines

Recognize event sequences that appear in a specific order: consecutive threshold crossings (debounced), rise-then-drop, start→run→stop workflows, out-of-order events. Standard SQL:2016 MATCH_RECOGNIZE, Flink-SQL-aligned, with four guards bounding edge memory.

-- Fire only after temperature crosses 50 three times in a row (debounce single-point jitter)
SELECT * FROM stream
MATCH_RECOGNIZE (
    ORDER BY ts
    MEASURES MATCH_NUMBER() AS mn, LAST(A.temp) AS peak
    ONE ROW PER MATCH
    PATTERN (A{3}) WITHIN '1h'
    DEFINE A AS temp > 50
)

Supports pattern variables with quantifiers (? * + {n}), alternation |, PERMUTE, navigation (PREV/NEXT/FIRST/LAST), aggregates, SUBSET, FINAL/RUNNING, WITHIN active expiry. See the CEP docs.

🔍 Analytic functions — CDC change detection & accumulation

Stateful computation across events on a windowless continuous stream — evaluated immediately on each event, with state retained across events.

-- CDC change detection: emit only when temperature changes, with the previous value
SELECT deviceId, temperature, lag(temperature) AS prev
FROM stream
WHERE had_changed(true, temperature)

-- Partitioned + cumulative: per-device state, running total since start
SELECT deviceId, acc_sum(score) OVER (PARTITION BY deviceId) AS total
FROM stream

OVER (PARTITION BY ... WHEN ...) controls partitioning and update conditions. See the analytic docs.

Which to use: compare adjacent events → analytic; ordered/sequence patterns → CEP; time-windowed stats → windowed aggregation + HAVING.

🪟 Windowed aggregation

Slice unbounded data into bounded segments for statistics, with 5 window types:

-- One tumbling window every 5 seconds, averaged per device
SELECT deviceId, AVG(temperature) AS avg_temp,
       window_start() AS start, window_end() AS end
FROM stream
GROUP BY deviceId, TumblingWindow('5s')
  • Tumbling TumblingWindow('5s'): fixed size, no overlap
  • Sliding SlidingWindow('30s','10s'): fixed size, slides by a step
  • Counting CountingWindow(100): by record count
  • Session SessionWindow('5m'): dynamic, by data activity
  • Global GLOBAL WINDOW TRIGGER WHEN ...: no time boundary, predicate-driven on the running aggregate, O(1) state per group
  • Built-in aggregates: MAX / MIN / AVG / SUM / COUNT / STDDEV / MEDIAN / PERCENTILE, with GROUP BY, HAVING
⏱ Event time & watermark

Two time semantics: event time (timestamps embedded in data) and processing time (system clock). Event-time windows use a watermark to handle out-of-order and late data:

SELECT deviceId, COUNT(*) AS cnt
FROM stream
GROUP BY deviceId, TumblingWindow('5m')
WITH (TIMESTAMP='eventTime', TIMEUNIT='ms',
      MAXOUTOFORDERNESS='5s',   -- tolerate 5s of out-of-order
      ALLOWEDLATENESS='2s',     -- accept 2s of late data after trigger
      IDLETIMEOUT='5s')         -- advance watermark on processing time after 5s idle
🧩 Nested fields

Dot notation for nested structures, index access for arrays:

SELECT device.info.name AS name, sensors[0].value AS v0
FROM stream WHERE device.info.type = 'temperature'
🔧 Custom functions

Register in one line, use immediately in SQL. Eight function types (math/string/conversion/datetime/aggregate/analytic/window/custom), addable and removable at runtime:

functions.RegisterCustomFunction("f2c", functions.TypeConversion,
    "Temperature", "Fahrenheit to Celsius", 1, 1,
    func(ctx *functions.FunctionContext, args []any) (any, error) {
        f, _ := functions.ConvertToFloat64(args[0])
        return (f - 32) * 5 / 9, nil
    })
// SELECT f2c(temperature) AS celsius FROM stream

Performance

x86 single-core / 128MB / v1.0.3, measured (BenchmarkGateway_* in test/e2e/stress_test.go):

Rule ns/op allocs msg/s
Filter 522 6 ~1.92M
Transform 1359 12 ~740K
Analytic + partition 2095 18 ~480K
  • 128MB holds 100k+ devices of partition state — memory is not the bottleneck, CPU throughput is.
  • Stability: no goroutine leaks, heap does not grow with load or partition count.
  • One rule saturating one core is the optimal edge-gateway usage; multi-core scales by running parallel independent instances (GOGC tuning gets near-linear).

ARM gateway figures are estimates derived from x86; measure on your target SoC before production. See the gateway capacity & performance benchmark.

Concepts

Two processing modes
  • Non-aggregation mode: no aggregate functions — each record is processed and emitted immediately, ultra-low latency. Data cleaning, real-time alerting, enrichment.
  • Aggregation mode: contains aggregate functions or GROUP BY — data goes into windows; aggregated results are emitted when windows trigger.
Windows

Stream data is unbounded and cannot be processed whole. Windows slice it into bounded segments: tumbling, sliding, counting, session, global (above).

Time semantics
  • Event time: when the data was actually generated (e.g. an event_time field). Windows are partitioned by timestamp; the watermark handles out-of-order/late data correctly — accurate, but with some latency.
  • Processing time: the system clock when data arrives (default). Low latency, but no handling of out-of-order/late data.
Feature Event time Processing time
Source Timestamp field in data System clock
Out-of-order / late Supported (watermark) Not supported
Accuracy Accurate May be inaccurate
Latency Higher Low
Config WITH (TIMESTAMP='field') Default (no WITH)

For deeper concepts (windows, watermark, late data) see the core concepts docs.

RuleGo integration

StreamSQL runs as RuleGo rule-chain nodes, tapping its 60+ components for any data source or third-party system, plus the rule engine:

  • streamTransform (x/streamTransform): non-aggregation SQL, row-by-row streaming transform
  • streamAggregator (x/streamAggregator): aggregation SQL, windowed aggregation
{
  "nodes": [{
    "id": "transform1", "type": "x/streamTransform",
    "configuration": { "sql": "SELECT deviceId, temperature*1.8+32 AS f FROM stream WHERE temperature>20" }
  }]
}

See the RuleGo integration docs.

Functions

60+ built-in functions: math, string, conversion, datetime, aggregate, analytic, window, and more. Function guide.

Contributing & Community

Issues and pull requests are welcome. Code should follow Go standards and include tests.

License

Apache License 2.0

Documentation

Overview

Package streamsql is a lightweight, SQL-based IoT edge stream processing engine.

StreamSQL provides efficient unbounded data stream processing and analysis capabilities, supporting multiple window types, aggregate functions, custom functions, and seamless integration with the RuleGo ecosystem.

Core Features

• Lightweight design - Pure in-memory operations, no external dependencies • SQL syntax support - Process stream data using familiar SQL syntax • Multiple window types - Sliding, tumbling, counting, and session windows • Event time and processing time - Support both time semantics for accurate stream processing • Watermark mechanism - Handle out-of-order and late-arriving data with configurable tolerance • Rich aggregate functions - MAX, MIN, AVG, SUM, STDDEV, MEDIAN, PERCENTILE, etc. • Plugin-based custom functions - Runtime dynamic registration, supports 8 function types • RuleGo ecosystem integration - Extend input/output sources using RuleGo components

Getting Started

Basic stream data processing: 

package main

import (
	"fmt"
	"math/rand"
	"time"
	"github.com/rulego/streamsql"
)

func main() {
	// Create StreamSQL instance
	ssql := streamsql.New()

	// Define SQL query - Calculate temperature average by device ID every 5 seconds
	sql := `SELECT deviceId,
		AVG(temperature) as avg_temp,
		MIN(humidity) as min_humidity,
		window_start() as start,
		window_end() as end
	FROM stream
	WHERE deviceId != 'device3'
	GROUP BY deviceId, TumblingWindow('5s')`

	// Execute SQL, create stream processing task
	err := ssql.Execute(sql)
	if err != nil {
		panic(err)
	}

	// Add result processing callback
	ssql.AddSink(func(result []map[string]any) {
		fmt.Printf("Aggregation result: %v\n", result)
	})

	// Simulate sending stream data
	go func() {
		ticker := time.NewTicker(1 * time.Second)
		defer ticker.Stop()

		for {
			select {
			case <-ticker.C:
				// Generate random device data
				data := map[string]any{
					"deviceId":    fmt.Sprintf("device%d", rand.Intn(3)+1),
					"temperature": 20.0 + rand.Float64()*10,
					"humidity":    50.0 + rand.Float64()*20,
				}
				ssql.Emit(data)
			}
		}
	}()

	// Run for 30 seconds
	time.Sleep(30 * time.Second)
}

Window Functions

StreamSQL supports multiple window types: 

// Tumbling window - Independent window every 5 seconds
SELECT AVG(temperature) FROM stream GROUP BY TumblingWindow('5s')

// Sliding window - 30-second window size, slides every 10 seconds
SELECT MAX(temperature) FROM stream GROUP BY SlidingWindow('30s', '10s')

// Counting window - One window per 100 records
SELECT COUNT(*) FROM stream GROUP BY CountingWindow(100)

// Session window - Automatically closes session after 5-minute timeout
SELECT user_id, COUNT(*) FROM stream GROUP BY user_id, SessionWindow('5m')

Event Time vs Processing Time

StreamSQL supports two time semantics for window processing:

## Processing Time (Default)

Processing time uses the system clock when data arrives. Windows are triggered based on data arrival time: 

// Processing time window (default)
SELECT COUNT(*) FROM stream GROUP BY TumblingWindow('5m')
// Windows are triggered every 5 minutes based on when data arrives

## Event Time

Event time uses timestamps embedded in the data itself. Windows are triggered based on event timestamps, allowing correct handling of out-of-order and late-arriving data: 

// Event time window - Use 'order_time' field as event timestamp
SELECT COUNT(*) as order_count
FROM stream
GROUP BY TumblingWindow('5m')
WITH (TIMESTAMP='order_time')

// Event time with integer timestamp (Unix milliseconds)
SELECT AVG(temperature) FROM stream
GROUP BY TumblingWindow('1m')
WITH (TIMESTAMP='event_time', TIMEUNIT='ms')

## Watermark and Late Data Handling

Event time windows use watermark mechanism to handle out-of-order and late data: 

// Configure max out-of-orderness (tolerate 5 seconds of out-of-order data)
SELECT COUNT(*) FROM stream
GROUP BY TumblingWindow('5m')
WITH (
	TIMESTAMP='order_time',
	MAXOUTOFORDERNESS='5s'  // Watermark = max(event_time) - 5s
)

// Configure allowed lateness (accept late data for 2 seconds after window closes)
SELECT COUNT(*) FROM stream
GROUP BY TumblingWindow('5m')
WITH (
	TIMESTAMP='order_time',
	ALLOWEDLATENESS='2s'  // Window stays open for 2s after trigger
)

// Combine both configurations
SELECT COUNT(*) FROM stream
GROUP BY TumblingWindow('5m')
WITH (
	TIMESTAMP='order_time',
	MAXOUTOFORDERNESS='5s',  // Tolerate 5s out-of-order before trigger
	ALLOWEDLATENESS='2s'     // Accept 2s late data after trigger
)

// Configure idle source mechanism (advance watermark based on processing time when data source is idle)
SELECT COUNT(*) FROM stream
GROUP BY TumblingWindow('5m')
WITH (
	TIMESTAMP='order_time',
	IDLETIMEOUT='5s'  // If no data arrives within 5s, watermark advances based on processing time
)

Key concepts: • MaxOutOfOrderness: Affects watermark calculation, delays window trigger to tolerate out-of-order data • AllowedLateness: Keeps window open after trigger to accept late data and update results • IdleTimeout: When data source is idle (no data arrives within timeout), watermark advances based on processing time to ensure windows can close • Watermark: Indicates that no events with timestamp less than watermark are expected

Custom Functions

StreamSQL supports plugin-based custom functions with runtime dynamic registration: 

// Register temperature conversion function
functions.RegisterCustomFunction(
	"fahrenheit_to_celsius",
	functions.TypeConversion,
	"Temperature conversion",
	"Fahrenheit to Celsius",
	1, 1,
	func(ctx *functions.FunctionContext, args []any) (any, error) {
		f, _ := functions.ConvertToFloat64(args[0])
		return (f - 32) * 5 / 9, nil
	},
)

// Use immediately in SQL
sql := `SELECT deviceId,
	AVG(fahrenheit_to_celsius(temperature)) as avg_celsius
FROM stream GROUP BY deviceId, TumblingWindow('5s')`

Supported custom function types: • TypeMath - Mathematical calculation functions • TypeString - String processing functions • TypeConversion - Type conversion functions • TypeDateTime - Date and time functions • TypeAggregation - Aggregate functions • TypeAnalytical - Analytical functions • TypeWindow - Window functions • TypeCustom - General custom functions

Log Configuration

StreamSQL provides flexible log configuration options: 

// Set log level
ssql := streamsql.New(streamsql.WithLogLevel(logger.DEBUG))

// Output to file
logFile, _ := os.OpenFile("app.log", os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0644)
ssql := streamsql.New(streamsql.WithLogOutput(logFile, logger.INFO))

// Disable logging (production environment)
ssql := streamsql.New(streamsql.WithDiscardLog())

RuleGo Integration

StreamSQL provides deep integration with the RuleGo rule engine through two dedicated components for stream data processing:

• streamTransform (x/streamTransform) - Stream transformer, handles non-aggregation SQL queries • streamAggregator (x/streamAggregator) - Stream aggregator, handles aggregation SQL queries

Basic integration example: 

package main

import (
	"github.com/rulego/rulego"
	"github.com/rulego/rulego/api/types"
	// Register StreamSQL components
	_ "github.com/rulego/rulego-components/external/streamsql"
)

func main() {
	// Rule chain configuration
	ruleChainJson := `{
		"ruleChain": {"id": "rule01"},
		"metadata": {
			"nodes": [{
				"id": "transform1",
				"type": "x/streamTransform",
				"configuration": {
					"sql": "SELECT deviceId, temperature * 1.8 + 32 as temp_f FROM stream WHERE temperature > 20"
				}
			}, {
				"id": "aggregator1",
				"type": "x/streamAggregator",
				"configuration": {
					"sql": "SELECT deviceId, AVG(temperature) as avg_temp FROM stream GROUP BY deviceId, TumblingWindow('5s')"
				}
			}],
			"connections": [{
				"fromId": "transform1",
				"toId": "aggregator1",
				"type": "Success"
			}]
		}
	}`

	// Create rule engine
	ruleEngine, _ := rulego.New("rule01", []byte(ruleChainJson))

	// Send data
	data := `{"deviceId":"sensor01","temperature":25.5}`
	msg := types.NewMsg(0, "TELEMETRY", types.JSON, types.NewMetadata(), data)
	ruleEngine.OnMsg(msg)
}

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type Option 

type Option func(*Streamsql)

Option defines the configuration option type for StreamSQL

func WithAnalyticMaxPartitions added in v1.0.3 

func WithAnalyticMaxPartitions(n int) Option

WithAnalyticMaxPartitions caps the number of PARTITION states kept per analytic function field (lag/had_changed/changed_col(s)/acc_*/latest with OVER(PARTITION BY...)). The least-recently-used partition is evicted above the cap. Only raise it when the partition key is genuinely high-cardinality (e.g. tens of thousands of devices behind one node) and memory allows: each partition holds one state plus its last result, ~150B for acc_* but up to several hundred bytes for changed_cols/had_changed('*') on wide rows. Default (n<=0) is 10000.

Note: evicting a partition resets its analytic state silently — acc_* counters restart from 0, lag returns its default, had_changed/changed_col treat the next row as the first. Size the cap above the peak active-partition count when cumulative accuracy matters.

func WithBufferSizes 

func WithBufferSizes(dataChannelSize, resultChannelSize, windowOutputSize int) Option

WithBufferSizes sets custom buffer sizes

func WithCustomPerformance 

func WithCustomPerformance(config types.PerformanceConfig) Option

WithCustomPerformance uses custom performance configuration

func WithDiscardLog 

func WithDiscardLog() Option

WithDiscardLog disables log output

func WithHighPerformance 

func WithHighPerformance() Option

WithHighPerformance uses high-performance configuration Suitable for scenarios requiring maximum throughput

func WithLogLevel 

func WithLogLevel(level logger.Level) Option

WithLogLevel sets the log level

func WithLogger added in v1.0.0 

func WithLogger(l logger.Logger) Option

WithLogger sets the per-instance logger used by this stream's pipeline. Each Streamsql instance logs only to its own logger (no global cross-talk between instances). Pass nil to keep the process default (logger.GetDefault).

func WithLowLatency 

func WithLowLatency() Option

WithLowLatency uses low-latency configuration Suitable for real-time interactive applications, minimizing latency

func WithMonitoring 

func WithMonitoring(updateInterval time.Duration, enableDetailedStats bool) Option

WithMonitoring enables detailed monitoring

func WithOverflowStrategy 

func WithOverflowStrategy(strategy string, blockTimeout time.Duration) Option

WithOverflowStrategy sets the overflow strategy

func WithSchema added in v1.0.0 

func WithSchema(s schema.Schema) Option

WithSchema registers an input-validation schema for this stream. Emit/EmitSync validate data against it and drop rows that fail (Emit logs+drops, EmitSync returns the error). Without WithSchema, Emit/EmitSync perform no validation (zero overhead). One Streamsql instance is one stream/query; for multiple streams, use multiple instances, each with its own schema.

func WithWorkerConfig 

func WithWorkerConfig(sinkPoolSize, sinkWorkerCount, maxRetryRoutines int) Option

WithWorkerConfig sets the worker pool configuration

type Streamsql 

type Streamsql struct {
	// contains filtered or unexported fields
}

Streamsql is the main interface for the StreamSQL streaming engine. It encapsulates core functionality including SQL parsing, stream processing, and window management.

Usage example: 

ssql := streamsql.New()
err := ssql.Execute("SELECT AVG(temperature) FROM stream GROUP BY TumblingWindow('5s')")
ssql.Emit(map[string]interface{}{"temperature": 25.5})

func New 

func New(options ...Option) *Streamsql

New creates a new StreamSQL instance. Supports configuration through optional Option parameters.

Parameters:

  • options: Variable configuration options for customizing StreamSQL behavior

Returns:

  • *Streamsql: Newly created StreamSQL instance

Examples: 

// Create default instance
ssql := streamsql.New()

// Create high performance instance
ssql := streamsql.New(streamsql.WithHighPerformance())

func (*Streamsql) AddSink 

func (s *Streamsql) AddSink(sink func([]map[string]interface{}))

AddSink directly adds result processing callback functions. Convenience wrapper for Stream().AddSink() for cleaner API calls.

Parameters:

  • sink: Result processing function, receives []map[string]interface{} type result data

Examples: 

// Directly add result processing
ssql.AddSink(func(results []map[string]interface{}) {
    fmt.Printf("Processing results: %v\n", results)
})

// Add multiple processors
ssql.AddSink(func(results []map[string]interface{}) {
    // Save to database
    saveToDatabase(results)
})
ssql.AddSink(func(results []map[string]interface{}) {
    // Send to message queue
    sendToQueue(results)
})

func (*Streamsql) AddSyncSink added in v0.10.6 

func (s *Streamsql) AddSyncSink(sink func([]map[string]interface{}))

AddSyncSink directly adds synchronous result processing callback functions. Convenience wrapper for Stream().AddSyncSink() for cleaner API calls.

Parameters:

  • sink: Result processing function, receives []map[string]interface{} type result data

Note: Sync sinks are executed sequentially in the result processing goroutine. Use this when order of execution matters.

func (*Streamsql) Emit 

func (s *Streamsql) Emit(data map[string]interface{})

Emit adds data to the stream processing pipeline. Accepts type-safe map[string]interface{} format data.

Parameters:

  • data: Data to add, must be map[string]interface{} type

Examples: 

// Add device data
ssql.Emit(map[string]interface{}{
    "deviceId": "sensor001",
    "temperature": 25.5,
    "humidity": 60.0,
    "timestamp": time.Now(),
})

// Add user behavior data
ssql.Emit(map[string]interface{}{
    "userId": "user123",
    "action": "click",
    "page": "/home",
})

func (*Streamsql) EmitSync 

func (s *Streamsql) EmitSync(data map[string]interface{}) (map[string]interface{}, error)

EmitSync processes data synchronously, returning results immediately. Only applicable for non-aggregation queries, aggregation queries will return an error. Accepts type-safe map[string]interface{} format data.

Parameters:

  • data: Data to process, must be map[string]interface{} type

Returns:

  • map[string]interface{}: Processed result data, returns nil if filter conditions don't match
  • error: Processing error

Examples: 

result, err := ssql.EmitSync(map[string]interface{}{
    "deviceId": "sensor001",
    "temperature": 25.5,
})
if err != nil {
    log.Printf("processing error: %v", err)
} else if result != nil {
    // Use processed result immediately (result is map[string]interface{} type)
    fmt.Printf("Processing result: %v\n", result)
}

func (*Streamsql) Execute 

func (s *Streamsql) Execute(sql string) error

Execute parses and executes SQL queries, creating corresponding stream processing pipelines. This is the core method of StreamSQL, responsible for converting SQL into actual stream processing logic.

Supported SQL syntax:

  • SELECT clause: Select fields and aggregate functions
  • FROM clause: Specify data source (usually 'stream')
  • WHERE clause: Data filtering conditions
  • GROUP BY clause: Grouping fields and window functions
  • HAVING clause: Aggregate result filtering
  • LIMIT clause: Limit result count
  • DISTINCT: Result deduplication

Window functions:

  • TumblingWindow('5s'): Tumbling window
  • SlidingWindow('30s', '10s'): Sliding window
  • CountingWindow(100): Counting window
  • SessionWindow('5m'): Session window

Parameters:

  • sql: SQL query statement to execute

Returns:

  • error: Returns error if SQL parsing or execution fails

Examples: 

// Basic aggregation query
err := ssql.Execute("SELECT deviceId, AVG(temperature) FROM stream GROUP BY deviceId, TumblingWindow('5s')")

// Query with filtering conditions
err := ssql.Execute("SELECT * FROM stream WHERE temperature > 30")

// Complex window aggregation
err := ssql.Execute(`
    SELECT deviceId,
           AVG(temperature) as avg_temp,
           MAX(humidity) as max_humidity
    FROM stream
    WHERE deviceId != 'test'
    GROUP BY deviceId, SlidingWindow('1m', '30s')
    HAVING avg_temp > 25
    LIMIT 100
`)

func (*Streamsql) GetDetailedStats 

func (s *Streamsql) GetDetailedStats() map[string]interface{}

GetDetailedStats returns detailed performance statistics

func (*Streamsql) GetStats 

func (s *Streamsql) GetStats() map[string]int64

GetStats returns stream processing statistics

func (*Streamsql) IsAggregationQuery 

func (s *Streamsql) IsAggregationQuery() bool

IsAggregationQuery checks if the current query is an aggregation query

func (*Streamsql) IsCEPQuery added in v1.1.0 

func (s *Streamsql) IsCEPQuery() bool

IsCEPQuery reports whether the current query runs the MATCH_RECOGNIZE (CEP) path. Components use it to route CEP queries to a dedicated node (Emit+AddSink), since CEP — like aggregation — does not support EmitSync.

func (*Streamsql) Metrics added in v1.0.0 

func (s *Streamsql) Metrics() *metrics.Registry

Metrics returns the underlying metrics registry, or nil before Execute. Callers can inspect named counters/gauges/histograms directly via the registry.

func (*Streamsql) PrintTable added in v0.10.1 

func (s *Streamsql) PrintTable()

PrintTable prints results to console in table format, similar to database output. Displays column names first, then data rows.

Supported data formats:

  • []map[string]interface{}: Multiple rows
  • map[string]interface{}: Single row
  • Other types: Direct print

Example: 

// Print results in table format
ssql.PrintTable()

// Output format:
// +--------+----------+
// | device | max_temp |
// +--------+----------+
// | aa     | 30.0     |
// | bb     | 22.0     |
// +--------+----------+

func (*Streamsql) RegisterTable added in v1.0.0 

func (s *Streamsql) RegisterTable(name string, rows []map[string]interface{}, keyFields ...string) (*stream.MemoryTableSource, error)

RegisterTable registers an in-memory metadata table for stream-table JOIN.

The index key is auto-derived from the JOIN ON clause's table-side field(s) when keyFields is omitted, so callers do not redeclare what ON already states (single or composite key both auto-derived). Pass keyFields explicitly to override. Returns the source so callers can Upsert/Delete rows incrementally.

Must be called after Execute. Examples: 

// Auto-derive key from ON (recommended): ON deviceId = m.deviceId
ssql.RegisterTable("meta", rows)
// Explicit composite key:
ssql.RegisterTable("meta", rows, "deviceId", "tenant")

func (*Streamsql) RegisterTableSource added in v1.0.0 

func (s *Streamsql) RegisterTableSource(src stream.TableSource) error

RegisterTableSource registers a custom table source (file/DB/Redis/HTTP). The implementation owns data loading, refresh, and cleanup; Lookup must be concurrency-safe. Must be called after Execute.

func (*Streamsql) SchemaDropped added in v1.0.0 

func (s *Streamsql) SchemaDropped() int64

SchemaDropped returns the count of rows dropped by schema validation.

func (*Streamsql) Stop 

func (s *Streamsql) Stop()

Stop stops the stream processor and releases related resources. After calling this method, the stream processor will stop receiving and processing new data.

Recommended to call this method for cleanup before application exit: 

defer ssql.Stop()

Note: StreamSQL instance cannot be restarted after stopping, create a new instance.

func (*Streamsql) Stream 

func (s *Streamsql) Stream() *stream.Stream

Stream returns the underlying stream processor instance. Provides access to lower-level stream processing functionality.

Returns:

  • *stream.Stream: Underlying stream processor instance, returns nil if SQL not executed

Common use cases:

  • Add result processing callbacks
  • Get result channel
  • Manually control stream processing lifecycle

Examples: 

// Add result processing callback
ssql.Stream().AddSink(func(results []map[string]interface{}) {
    fmt.Printf("Processing results: %v\n", results)
})

// Get result channel
resultChan := ssql.Stream().GetResultsChan()
go func() {
    for result := range resultChan {
        // Process result
    }
}()

func (*Streamsql) ToChannel 

func (s *Streamsql) ToChannel() <-chan []map[string]interface{}

ToChannel converts query results to channel output Returns a read-only channel for receiving query results

Notes:

  • Consumer must continuously read from channel to prevent stream processing blocking
  • Channel transmits batch result data

func (*Streamsql) TriggerWindow added in v1.0.0 

func (s *Streamsql) TriggerWindow()

TriggerWindow manually triggers the current window to emit immediately, bypassing its normal time/count trigger. Intended for tests that need a window to fire deterministically, and as an explicit flush hook.

func (*Streamsql) UpsertTable added in v1.0.0 

func (s *Streamsql) UpsertTable(name string, row map[string]interface{}) error

UpsertTable adds or replaces a row in a previously registered in-memory table. Only affects rows emitted after the call (tables are snapshots).

Source Files

View all Source files

Directories

Path Synopsis
Package aggregator provides data aggregation functionality for StreamSQL.
 Package aggregator provides data aggregation functionality for StreamSQL.
Package cep 实现 MATCH_RECOGNIZE 模式识别（SQL:2016）。
 Package cep 实现 MATCH_RECOGNIZE 模式识别（SQL:2016）。
Package condition provides condition evaluation functionality for StreamSQL.
 Package condition provides condition evaluation functionality for StreamSQL.
examples
advanced-functions command
complex-nested-access command
comprehensive-test command
custom-functions-demo command
function-integration-demo command
nested-field-examples command
non-aggregation command
null-comparison-examples command
simple-custom-functions command
table_print_demo command
Package expr provides expression parsing and evaluation capabilities for StreamSQL.
 Package expr provides expression parsing and evaluation capabilities for StreamSQL.
Package functions provides a comprehensive function registry and execution framework for StreamSQL.
 Package functions provides a comprehensive function registry and execution framework for StreamSQL.
Package logger provides logging functionality for StreamSQL.
 Package logger provides logging functionality for StreamSQL.
Package rsql provides SQL parsing and analysis capabilities for StreamSQL.
 Package rsql provides SQL parsing and analysis capabilities for StreamSQL.
Package schema provides a runtime registry for typed record schemas and validates incoming maps against the declared field definitions.
 Package schema provides a runtime registry for typed record schemas and validates incoming maps against the declared field definitions.
Package stream provides the core stream processing engine for StreamSQL.
 Package stream provides the core stream processing engine for StreamSQL.
Package types provides core type definitions and data structures for StreamSQL.
 Package types provides core type definitions and data structures for StreamSQL.
utils
cast
fieldpath
reflectutil
table
timex
Package window provides windowing functionality for StreamSQL stream processing.
 Package window provides windowing functionality for StreamSQL stream processing.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.