Merge pull request #18 from rulego/dev

Dev

aggregator/group_aggregator.go

@@ -50,12 +50,12 @@ func NewGroupAggregator(groupFields []string, aggregationFields []AggregationFie
 	aggregators := make(map[string]AggregatorFunction)
 
 	// 为每个聚合字段创建聚合器
-	for _, field := range aggregationFields {
-		if field.OutputAlias == "" {
+	for i := range aggregationFields {
+		if aggregationFields[i].OutputAlias == "" {
 			// 如果没有指定别名，使用输入字段名
-			field.OutputAlias = field.InputField
+			aggregationFields[i].OutputAlias = aggregationFields[i].InputField
 		}
-		aggregators[field.OutputAlias] = CreateBuiltinAggregator(field.AggregateType)
+		aggregators[aggregationFields[i].OutputAlias] = CreateBuiltinAggregator(aggregationFields[i].AggregateType)
 	}
 
 	return &GroupAggregator{

functions/expr_bridge.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"strconv"
 	"strings"
+	"sync"
 
 	"github.com/expr-lang/expr"
 	"github.com/expr-lang/expr/vm"
@@ -15,6 +16,7 @@ type ExprBridge struct {
 	streamSQLFunctions map[string]Function
 	exprProgram        *vm.Program
 	exprEnv            map[string]interface{}
+	mutex              sync.RWMutex // 添加读写锁保护并发访问
 }
 
 // NewExprBridge 创建新的表达式桥接器
@@ -27,6 +29,9 @@ func NewExprBridge() *ExprBridge {
 
 // RegisterStreamSQLFunctionsToExpr 将StreamSQL函数注册到expr环境中
 func (bridge *ExprBridge) RegisterStreamSQLFunctionsToExpr() []expr.Option {
+	bridge.mutex.Lock()
+	defer bridge.mutex.Unlock()
+
 	options := make([]expr.Option, 0)
 
 	// 将所有StreamSQL函数注册到expr环境
@@ -58,6 +63,9 @@ func (bridge *ExprBridge) RegisterStreamSQLFunctionsToExpr() []expr.Option {
 
 // CreateEnhancedExprEnvironment 创建增强的expr执行环境
 func (bridge *ExprBridge) CreateEnhancedExprEnvironment(data map[string]interface{}) map[string]interface{} {
+	bridge.mutex.RLock()
+	defer bridge.mutex.RUnlock()
+
 	// 合并数据和函数环境
 	env := make(map[string]interface{})
 
@@ -323,6 +331,9 @@ func (bridge *ExprBridge) toFloat64(val interface{}) (float64, error) {
 
 // GetFunctionInfo 获取函数信息，统一两个系统的函数
 func (bridge *ExprBridge) GetFunctionInfo() map[string]interface{} {
+	bridge.mutex.RLock()
+	defer bridge.mutex.RUnlock()
+
 	info := make(map[string]interface{})
 
 	// StreamSQL函数信息
@@ -394,6 +405,9 @@ func (bridge *ExprBridge) GetFunctionInfo() map[string]interface{} {
 
 // ResolveFunction 解析函数调用，优先使用StreamSQL函数
 func (bridge *ExprBridge) ResolveFunction(name string) (interface{}, bool, string) {
+	bridge.mutex.RLock()
+	defer bridge.mutex.RUnlock()
+
 	// 进行大小写不敏感的查找
 	lowerName := strings.ToLower(name)
 
@@ -445,9 +459,23 @@ func (bridge *ExprBridge) IsExprLangFunction(name string) bool {
 
 // 全局桥接器实例
 var globalBridge *ExprBridge
+var globalBridgeMutex sync.RWMutex
 
 // GetExprBridge 获取全局桥接器实例
 func GetExprBridge() *ExprBridge {
+	// 首先使用读锁检查是否已初始化
+	globalBridgeMutex.RLock()
+	if globalBridge != nil {
+		defer globalBridgeMutex.RUnlock()
+		return globalBridge
+	}
+	globalBridgeMutex.RUnlock()
+
+	// 使用写锁进行初始化
+	globalBridgeMutex.Lock()
+	defer globalBridgeMutex.Unlock()
+
+	// 双重检查模式，防止并发初始化
 	if globalBridge == nil {
 		globalBridge = NewExprBridge()
 	}

window/sliding_window.go

@@ -212,18 +212,13 @@ func (sw *SlidingWindow) Trigger() {
 	sw.data = newData
 	sw.currentSlot = next
 
-	// 非阻塞发送到输出通道
-	sw.sendResultNonBlocking(resultData)
-}
-
-// sendResultNonBlocking 非阻塞地发送结果到输出通道
-func (sw *SlidingWindow) sendResultNonBlocking(resultData []types.Row) {
+	// 非阻塞发送到输出通道并直接更新统计信息（已在锁内）
 	select {
 	case sw.outputChan <- resultData:
-		// 成功发送
+		// 成功发送，更新统计信息（已在锁内）
 		sw.sentCount++
 	default:
-		// 通道已满，丢弃结果
+		// 通道已满，丢弃结果并更新统计信息（已在锁内）
 		sw.droppedCount++
 	}
 }

window/tumbling_window.go

@@ -209,18 +209,13 @@ func (tw *TumblingWindow) Trigger() {
 	tw.data = newData
 	tw.currentSlot = next
 
-	// 非阻塞发送到输出通道
-	tw.sendResultNonBlocking(resultData)
-}
-
-// sendResultNonBlocking 非阻塞地发送结果到输出通道
-func (tw *TumblingWindow) sendResultNonBlocking(resultData []types.Row) {
+	// 非阻塞发送到输出通道并更新统计信息
 	select {
 	case tw.outputChan <- resultData:
-		// 成功发送
+		// 成功发送，更新统计信息（已在锁内）
 		tw.sentCount++
 	default:
-		// 通道已满，丢弃结果（可选：记录日志或触发告警）
+		// 通道已满，丢弃结果并更新统计信息（已在锁内）
 		tw.droppedCount++
 		// 可选：在这里添加日志记录
 		// log.Printf("Window output channel full, dropped result with %d rows", len(resultData))