stream

package
v1.0.2 Latest Latest
Warning

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

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

Documentation

Overview

Package stream provides the core stream processing engine for StreamSQL.

This package implements the central stream processing pipeline that orchestrates data flow, window management, aggregation, filtering, and result generation. It serves as the execution engine that brings together all StreamSQL components into a cohesive streaming system.

Core Features

• Real-time Stream Processing - High-throughput, low-latency data stream processing • Window Management - Integration with all window types (tumbling, sliding, counting, session) • Aggregation Engine - Efficient aggregation with incremental computation • Filtering Pipeline - Multi-stage filtering with WHERE and HAVING clause support • Performance Modes - Configurable performance profiles for different use cases • Metrics and Monitoring - Comprehensive performance metrics and health monitoring • Persistence Support - Optional data persistence for reliability and recovery • Backpressure Handling - Intelligent backpressure management and overflow strategies

Stream Architecture

The stream processing pipeline consists of several key components: 

type Stream struct {
	dataChan       chan map[string]any  // Input data channel
	filter         condition.Condition          // WHERE clause filter
	Window         window.Window                // Window manager
	aggregator     aggregator.Aggregator        // Aggregation engine
	config         types.Config                 // Stream configuration
	sinks          []func([]map[string]any) // Result processors
	resultChan     chan []map[string]any // Result channel

	dataStrategy   DataProcessingStrategy       // Data processing strategy
}

Performance Modes

Configurable performance profiles for different scenarios: 

// High Performance Mode
// - Optimized for maximum throughput
// - Larger buffer sizes
// - Batch processing optimization
stream := NewStreamWithHighPerformance(config)

// Low Latency Mode
// - Optimized for minimal processing delay
// - Smaller buffer sizes
// - Immediate processing
stream := NewStreamWithLowLatency(config)

// Custom Performance Mode
// - User-defined performance parameters
customConfig := &PerformanceConfig{
	BufferSize:     1000,
	BatchSize:      50,
	FlushInterval:  time.Second,
	WorkerPoolSize: 4,
}
stream := NewStreamWithCustomPerformance(config, *customConfig)

Data Processing Pipeline

Multi-stage processing pipeline with optimized data flow:

  1. Data Ingestion ├── Input validation and type checking ├── Timestamp extraction and normalization └── Initial data transformation

  2. Filtering (WHERE clause) ├── Field-based filtering ├── Expression evaluation └── Early data rejection

  3. Window Processing ├── Window assignment ├── Data buffering └── Window trigger management

  4. Aggregation ├── Group-by processing ├── Aggregate function execution └── Incremental computation

  5. Post-Aggregation Filtering (HAVING clause) ├── Aggregate result filtering ├── Complex condition evaluation └── Final result validation

  6. Result Generation ├── Field projection ├── Alias application └── Output formatting

Window Integration

Seamless integration with all window types: 

// Tumbling Windows - Non-overlapping time-based windows
config.WindowConfig = WindowConfig{
	Type: "tumbling",
	Params: map[string]any{
		"size": "5s",
	},
}

// Sliding Windows - Overlapping time-based windows
config.WindowConfig = WindowConfig{
	Type: "sliding",
	Params: map[string]any{
		"size": "30s",
		"slide": "10s",
	},
}

// Counting Windows - Count-based windows
config.WindowConfig = WindowConfig{
	Type: "counting",
	Params: map[string]any{
		"count": 100,
	},
}

// Session Windows - Activity-based windows
config.WindowConfig = WindowConfig{
	Type: "session",
	Params: map[string]any{
		"timeout": "5m",
		"groupBy": "user_id",
	},
}

Metrics and Monitoring

Comprehensive performance monitoring: 

type MetricsManager struct {
	processedCount    int64     // Total processed records
	filteredCount     int64     // Filtered out records
	aggregatedCount   int64     // Aggregated records
	errorCount        int64     // Processing errors
	processingTime    time.Duration // Average processing time
	throughput        float64   // Records per second
	memoryUsage       int64     // Memory consumption
	bufferUtilization float64   // Buffer usage percentage
}

// Get basic statistics
stats := stream.GetStats()
fmt.Printf("Processed: %d, Errors: %d\n", stats["processed"], stats["errors"])

// Get detailed performance metrics
detailed := stream.GetDetailedStats()
fmt.Printf("Throughput: %.2f records/sec\n", detailed["throughput"])
fmt.Printf("Memory Usage: %d bytes\n", detailed["memory_usage"])

Backpressure Management

Intelligent handling of system overload: 

// Overflow strategies
const (
	OverflowStrategyDrop     = "drop"     // Drop oldest data
	OverflowStrategyBlock    = "block"    // Block new data
	OverflowStrategySpill    = "spill"    // Spill to disk
	OverflowStrategyCompress = "compress" // Compress data
)

// Configure backpressure handling
config.PerformanceConfig.OverflowStrategy = OverflowStrategySpill
config.PerformanceConfig.BufferSize = 10000
config.PerformanceConfig.HighWaterMark = 0.8

Usage Examples

Basic stream processing: 

// Create stream with default configuration
stream, err := NewStream(config)
if err != nil {
	log.Fatal(err)
}

// Register result handler
stream.AddSink(func(results []map[string]any) {
	fmt.Printf("Results: %v\n", results)
})

// Start processing
stream.Start()

// Send data
stream.Emit(map[string]any{
	"device_id":   "sensor001",
	"temperature": 25.5,
	"timestamp":   time.Now(),
})

High-performance stream processing: 

// Create high-performance stream
stream, err := NewStreamWithHighPerformance(config)

// Configure for maximum throughput
stream.SetBufferSize(50000)
stream.SetBatchSize(1000)
stream.SetWorkerPoolSize(8)

// Enable metrics monitoring
stream.EnableMetrics(true)

// Process data in batches
for _, batch := range dataBatches {
	stream.EmitBatch(batch)
}

Synchronous processing for non-aggregation queries: 

// Process single record synchronously
result, err := stream.ProcessSync(data)
if err != nil {
	log.Printf("Processing error: %v", err)
} else if result != nil {
	fmt.Printf("Immediate result: %v\n", result)
}

Integration

Central integration point for all StreamSQL components:

• RSQL package - Configuration parsing and application • Window package - Window lifecycle management • Aggregator package - Aggregation execution • Functions package - Function execution in expressions • Condition package - Filter condition evaluation • Types package - Data type handling and configuration • Logger package - Comprehensive logging and debugging

Index

Constants

View Source 
const (
	InputCount         = "input_count"
	OutputCount        = "output_count"
	InputDroppedCount  = "input_dropped_count"
	OutputDroppedCount = "output_dropped_count"
	DroppedCount       = "dropped_count"
	DataChanLen        = "data_chan_len"
	DataChanCap        = "data_chan_cap"
	ResultChanLen      = "result_chan_len"
	ResultChanCap      = "result_chan_cap"
	SinkPoolLen        = "sink_pool_len"
	SinkPoolCap        = "sink_pool_cap"
	ActiveRetries      = "active_retries"
	Expanding          = "expanding"
)

Statistic field keys returned by GetStats.

View Source 
const (
	BasicStats       = "basic_stats"
	DataChanUsage    = "data_chan_usage"
	ResultChanUsage  = "result_chan_usage"
	SinkPoolUsage    = "sink_pool_usage"
	ProcessRate      = "process_rate"
	DropRate         = "drop_rate"
	PerformanceLevel = "performance_level"
)

Detailed statistics field keys returned by GetDetailedStats.

View Source 
const (
	StrategyDrop   = "drop"   // Drop strategy
	StrategyBlock  = "block"  // Blocking strategy
	StrategyExpand = "expand" // Dynamic strategy
)

Overflow strategy constants

View Source 
const (
	WindowStartField = "window_start"
	WindowEndField   = "window_end"
)

Window related constants

View Source 
const (
	PerformanceLevelCritical     = "CRITICAL"
	PerformanceLevelWarning      = "WARNING"
	PerformanceLevelHighLoad     = "HIGH_LOAD"
	PerformanceLevelModerateLoad = "MODERATE_LOAD"
	PerformanceLevelOptimal      = "OPTIMAL"
)

Performance level constants

View Source 
const (
	PerformanceConfigKey = "performanceConfig"
)
View Source 
const (
	SQLKeywordCase = "CASE"
)

SQL keyword constants

Variables

This section is empty.

Functions

func AssessPerformanceLevel added in v0.10.2 

func AssessPerformanceLevel(dataUsage, dropRate float64) string

AssessPerformanceLevel maps data usage and drop rate to a performance level.

Types

type BlockingStrategy added in v0.10.2 

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

BlockingStrategy blocking strategy implementation

func NewBlockingStrategy added in v0.10.2 

func NewBlockingStrategy() *BlockingStrategy

NewBlockingStrategy creates blocking strategy instance

func (*BlockingStrategy) GetStrategyName added in v0.10.2 

func (bs *BlockingStrategy) GetStrategyName() string

GetStrategyName gets strategy name

func (*BlockingStrategy) Init added in v0.10.2 

func (bs *BlockingStrategy) Init(stream *Stream, config types.PerformanceConfig) error

Init initializes blocking strategy

func (*BlockingStrategy) ProcessData added in v0.10.2 

func (bs *BlockingStrategy) ProcessData(data map[string]any)

ProcessData implements blocking mode data processing. blockingTimeout <= 0: block forever (pure backpressure, never drops). blockingTimeout > 0: block up to the timeout, then drop so a slow consumer cannot hang the producer — bounded block is the explicit contract.

func (*BlockingStrategy) Stop added in v0.10.2 

func (bs *BlockingStrategy) Stop() error

Stop stops and cleans up blocking strategy resources

type DataHandler added in v0.10.2 

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

DataHandler handles data processing for different strategies

func NewDataHandler added in v0.10.2 

func NewDataHandler(stream *Stream) *DataHandler

NewDataHandler creates a new data handler

type DataProcessingStrategy added in v0.10.2 

type DataProcessingStrategy interface {
	// ProcessData core method for processing data
	// Parameters:
	//   - data: data to process, must be map[string]any type
	ProcessData(data map[string]any)

	// GetStrategyName gets strategy name
	GetStrategyName() string

	// Init initializes strategy
	// Parameters:
	//   - stream: Stream instance reference
	//   - config: performance configuration
	Init(stream *Stream, config types.PerformanceConfig) error

	// Stop stops and cleans up resources
	Stop() error
}

DataProcessingStrategy data processing strategy interface Defines unified interface for different overflow strategies, providing better extensibility and maintainability

type DataProcessor added in v0.10.2 

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

DataProcessor data processor responsible for processing data streams

func NewDataProcessor added in v0.10.2 

func NewDataProcessor(stream *Stream) *DataProcessor

NewDataProcessor creates a data processor

func (*DataProcessor) Process added in v0.10.2 

func (dp *DataProcessor) Process()

Process main processing loop

type DropStrategy added in v0.10.2 

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

DropStrategy drop strategy implementation

func NewDropStrategy added in v0.10.2 

func NewDropStrategy() *DropStrategy

NewDropStrategy creates drop strategy instance

func (*DropStrategy) GetStrategyName added in v0.10.2 

func (ds *DropStrategy) GetStrategyName() string

GetStrategyName gets strategy name

func (*DropStrategy) Init added in v0.10.2 

func (ds *DropStrategy) Init(stream *Stream, config types.PerformanceConfig) error

Init initializes drop strategy

func (*DropStrategy) ProcessData added in v0.10.2 

func (ds *DropStrategy) ProcessData(data map[string]any)

ProcessData implements drop mode: non-blocking send, a few short retries to absorb micro-bursts, then drop.

func (*DropStrategy) Stop added in v0.10.2 

func (ds *DropStrategy) Stop() error

Stop stops and cleans up drop strategy resources

type ExpansionStrategy added in v0.10.2 

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

ExpansionStrategy expansion strategy implementation

func NewExpansionStrategy added in v0.10.2 

func NewExpansionStrategy() *ExpansionStrategy

NewExpansionStrategy creates expansion strategy instance

func (*ExpansionStrategy) GetStrategyName added in v0.10.2 

func (es *ExpansionStrategy) GetStrategyName() string

GetStrategyName gets strategy name

func (*ExpansionStrategy) Init added in v0.10.2 

func (es *ExpansionStrategy) Init(stream *Stream, config types.PerformanceConfig) error

Init initializes expansion strategy

func (*ExpansionStrategy) ProcessData added in v0.10.2 

func (es *ExpansionStrategy) ProcessData(data map[string]any)

ProcessData implements expansion mode: non-blocking send, expand on full, retry, then a few short waits for the consumer to drain before dropping. It never blocks on a cached channel reference — under concurrent expansion such a reference can be swapped out and strand the send (and hang the caller).

func (*ExpansionStrategy) Stop added in v0.10.2 

func (es *ExpansionStrategy) Stop() error

Stop stops and cleans up expansion strategy resources

type MemoryTableSource added in v1.0.0 

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

MemoryTableSource is an in-memory table indexed by one or more key fields. It is purely push-based (no background goroutine): callers mutate it via Upsert/Delete, or rebuild it wholesale by registering a new source.

func NewMemoryTableSource added in v1.0.0 

func NewMemoryTableSource(name string, keyFields []string, rows []map[string]any) *MemoryTableSource

NewMemoryTableSource builds an in-memory table from rows indexed by keyFields. keyFields must be in the same order as the JOIN ON table-side fields.

func (*MemoryTableSource) Close added in v1.0.0 

func (m *MemoryTableSource) Close() error

Close is a no-op for the in-memory source.

func (*MemoryTableSource) Delete added in v1.0.0 

func (m *MemoryTableSource) Delete(key any)

Delete removes the row whose key-field values match key.

func (*MemoryTableSource) Init added in v1.0.0 

func (m *MemoryTableSource) Init() error

Init is a no-op for the in-memory source (data is supplied at construction).

func (*MemoryTableSource) KeyFields added in v1.0.0 

func (m *MemoryTableSource) KeyFields() []string

KeyFields returns the fields the table is indexed by.

func (*MemoryTableSource) Lookup added in v1.0.0 

func (m *MemoryTableSource) Lookup(key any) (map[string]any, bool)

Lookup returns the row matching key, or (nil, false). key is a []any of key-field values in indexed order, or a single value for single-key tables.

func (*MemoryTableSource) Name added in v1.0.0 

func (m *MemoryTableSource) Name() string

Name returns the table source name.

func (*MemoryTableSource) Upsert added in v1.0.0 

func (m *MemoryTableSource) Upsert(row map[string]any)

Upsert adds or replaces the row, keyed by its key-field values.

type ResultHandler added in v0.10.2 

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

ResultHandler handles result output and sink function calls

func NewResultHandler added in v0.10.2 

func NewResultHandler(stream *Stream) *ResultHandler

NewResultHandler creates a new result handler

type Sorter added in v1.0.0 

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

Sorter orders result rows by the configured ORDER BY keys. Sorting is applied per emit batch (windowed queries) — global ordering over an unbounded stream is not possible. Stability: equal-key rows keep their input order (sort.SliceStable).

func NewSorter added in v1.0.0 

func NewSorter(keys []types.OrderByField) *Sorter

NewSorter builds a Sorter for the given keys. A nil/empty slice is a no-op.

func (*Sorter) Sort added in v1.0.0 

func (s *Sorter) Sort(rows []map[string]any)

Sort sorts rows in place. It is a no-op when there are no keys or fewer than two rows.

type StrategyFactory added in v0.10.2 

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

StrategyFactory strategy factory Uses unified registration mechanism to manage all strategies (built-in and custom)

func NewStrategyFactory added in v0.10.2 

func NewStrategyFactory() *StrategyFactory

NewStrategyFactory creates strategy factory instance Automatically registers all built-in strategies

func (*StrategyFactory) CreateStrategy added in v0.10.2 

func (sf *StrategyFactory) CreateStrategy(strategyName string) (DataProcessingStrategy, error)

CreateStrategy creates corresponding strategy instance based on strategy name Parameters:

  • strategyName: strategy name

Returns:

  • DataProcessingStrategy: strategy instance
  • error: error information

func (*StrategyFactory) GetRegisteredStrategies added in v0.10.2 

func (sf *StrategyFactory) GetRegisteredStrategies() []string

GetRegisteredStrategies gets all registered strategy names Returns:

  • []string: strategy name list

func (*StrategyFactory) RegisterStrategy added in v0.10.2 

func (sf *StrategyFactory) RegisterStrategy(name string, constructor func() DataProcessingStrategy)

RegisterStrategy registers strategy to factory Parameters:

  • name: strategy name
  • constructor: strategy constructor function

func (*StrategyFactory) UnregisterStrategy added in v0.10.2 

func (sf *StrategyFactory) UnregisterStrategy(name string)

UnregisterStrategy unregisters strategy Parameters:

  • name: strategy name

type Stream 

type Stream struct {
	Window window.Window
	// contains filtered or unexported fields
}

func NewStream 

func NewStream(config types.Config) (*Stream, error)

NewStream creates Stream using unified configuration

func NewStreamWithCustomPerformance 

func NewStreamWithCustomPerformance(config types.Config, perfConfig types.PerformanceConfig) (*Stream, error)

NewStreamWithCustomPerformance creates Stream with custom performance configuration

func NewStreamWithHighPerformance 

func NewStreamWithHighPerformance(config types.Config) (*Stream, error)

NewStreamWithHighPerformance creates high-performance Stream

func NewStreamWithLowLatency 

func NewStreamWithLowLatency(config types.Config) (*Stream, error)

NewStreamWithLowLatency creates low-latency Stream

func (*Stream) AddSink 

func (s *Stream) AddSink(sink func([]map[string]any))

AddSink adds a sink function Parameters:

  • sink: result processing function that receives []map[string]any type result data

Note: Sinks are executed asynchronously in a worker pool, so execution order is NOT guaranteed. If you need strict ordering, use GetResultsChan() instead.

func (*Stream) AddSyncSink added in v0.10.6 

func (s *Stream) AddSyncSink(sink func([]map[string]any))

AddSyncSink adds a synchronous sink function Parameters:

  • sink: result processing function that receives []map[string]any type result data

Note: Sync sinks are executed sequentially in the result processing goroutine. They block subsequent processing, so they should be fast.

func (*Stream) Emit 

func (s *Stream) Emit(data map[string]any)

Emit adds data to stream processing pipeline Parameters:

  • data: data to be processed, must be map[string]any type

func (*Stream) GetDetailedStats 

func (s *Stream) GetDetailedStats() map[string]any

GetDetailedStats gets detailed performance statistics

func (*Stream) GetResultsChan 

func (s *Stream) GetResultsChan() <-chan []map[string]any

GetResultsChan gets the result channel

func (*Stream) GetStats 

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

GetStats gets stream processing statistics (thread-safe version)

func (*Stream) IsAggregationQuery 

func (s *Stream) IsAggregationQuery() bool

IsAggregationQuery checks if current stream is an aggregation query

func (*Stream) JoinKeyFields added in v1.0.0 

func (s *Stream) JoinKeyFields(table string) ([]string, error)

JoinKeyFields returns the table-side key fields for a table by looking up the JOIN config that references it. This lets RegisterTable auto-derive the index key from the ON clause instead of requiring the caller to redeclare it. Returns an error if no JOIN references the table.

func (*Stream) MetricsRegistry added in v1.0.0 

func (s *Stream) MetricsRegistry() *metrics.Registry

MetricsRegistry returns the stream's metrics registry.

func (*Stream) ProcessSync 

func (s *Stream) ProcessSync(data map[string]any) (map[string]any, error)

ProcessSync synchronously processes single data, returns result immediately Only applicable to non-aggregation queries, aggregation queries will return error Parameters:

  • data: data to be processed, must be map[string]any type

Returns:

  • map[string]any: processed result data, returns nil if doesn't match filter condition
  • error: processing error, returns error for aggregation queries

func (*Stream) RegisterFilter 

func (s *Stream) RegisterFilter(conditionStr string) error

RegisterFilter registers filter condition, supporting backtick identifiers, LIKE syntax and IS NULL syntax

func (*Stream) RegisterMemoryTable added in v1.0.0 

func (s *Stream) RegisterMemoryTable(name string, keyFields []string, rows []map[string]any) (*MemoryTableSource, error)

RegisterMemoryTable registers an in-memory table indexed by keyFields, for stream-table JOIN. keyFields order must match the JOIN ON table-side fields. Returns the source for incremental Upsert/Delete.

func (*Stream) RegisterTableSource added in v1.0.0 

func (s *Stream) RegisterTableSource(src TableSource) error

RegisterTableSource registers a custom table source for stream-table JOIN. The source's Init runs here (it may load data from a file/DB/Redis).

func (*Stream) ResetStats 

func (s *Stream) ResetStats()

ResetStats resets statistics information

func (*Stream) Start 

func (s *Stream) Start()

func (*Stream) Stop 

func (s *Stream) Stop()

Stop stops stream processing

func (*Stream) UpsertTableRow added in v1.0.0 

func (s *Stream) UpsertTableRow(name string, row map[string]any) error

UpsertTableRow adds or replaces a row in a registered memory table.

type StreamFactory added in v0.10.2 

type StreamFactory struct{}

StreamFactory Stream factory responsible for creating different types of Streams

func NewStreamFactory added in v0.10.2 

func NewStreamFactory() *StreamFactory

NewStreamFactory creates Stream factory

func (*StreamFactory) CreateCustomPerformanceStream added in v0.10.2 

func (sf *StreamFactory) CreateCustomPerformanceStream(config types.Config, perfConfig types.PerformanceConfig) (*Stream, error)

CreateCustomPerformanceStream creates Stream with custom performance configuration

func (*StreamFactory) CreateHighPerformanceStream added in v0.10.2 

func (sf *StreamFactory) CreateHighPerformanceStream(config types.Config) (*Stream, error)

CreateHighPerformanceStream creates high-performance Stream

func (*StreamFactory) CreateLowLatencyStream added in v0.10.2 

func (sf *StreamFactory) CreateLowLatencyStream(config types.Config) (*Stream, error)

CreateLowLatencyStream creates low-latency Stream

func (*StreamFactory) CreateStream added in v0.10.2 

func (sf *StreamFactory) CreateStream(config types.Config) (*Stream, error)

CreateStream creates Stream using unified configuration

type TableSource added in v1.0.0 

type TableSource interface {
	Name() string
	Lookup(key any) (map[string]any, bool)
	Init() error
	Close() error
}

TableSource backs a stream-table JOIN (v0.5: metadata enrichment). Implementations own the data lifecycle (load/refresh/cleanup) and MUST make Lookup safe for concurrent use, since it runs on the per-row hot path.

Lookup receives the join key built by the engine: a []any of the stream-side key values, in the same order as the ON clause's table-side fields (which must match the source's indexed key fields). For a single-key JOIN the slice has one element.

Source Files

View all Source files

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.