streamsql_test.go

package streamsql

import (
	"context"
	"fmt"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/rulego/streamsql/utils/cast"

	"math/rand"

	"github.com/rulego/streamsql/functions"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
)

// TestStreamData 测试 StreamSQL 的流式数据处理功能
// 这个测试演示了 StreamSQL 的完整工作流程：从创建实例到数据处理再到结果验证
func TestStreamData(t *testing.T) {
	// 步骤1: 创建 StreamSQL 实例
	// StreamSQL 是流式 SQL 处理引擎的核心组件，负责管理整个流处理生命周期
	ssql := New()
	// 确保测试结束时停止流处理，释放资源
	defer ssql.Stop()

	// 步骤2: 定义流式 SQL 查询语句
	// 这个 SQL 语句展示了 StreamSQL 的核心功能：
	// - SELECT: 选择要输出的字段和聚合函数
	// - FROM stream: 指定数据源为流数据
	// - WHERE: 过滤条件，排除 device3 的数据
	// - GROUP BY: 按设备ID分组，配合滚动窗口进行聚合
	// - TumblingWindow('5s'): 5秒滚动窗口，每5秒触发一次计算
	// - avg(), min(): 聚合函数，计算平均值和最小值
	// - window_start(), window_end(): 窗口函数，获取窗口的开始和结束时间
	rsql := "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')"

	// 步骤3: 执行 SQL 语句，启动流式分析任务
	// Execute 方法会解析 SQL、构建执行计划、初始化窗口管理器和聚合器
	err := ssql.Execute(rsql)
	if err != nil {
		panic(err)
	}

	// 步骤4: 设置测试环境和并发控制
	var wg sync.WaitGroup
	wg.Add(1)
	// 设置30秒测试超时时间，防止测试无限运行
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	// 步骤5: 启动数据生产者协程
	// 模拟实时数据流，持续向 StreamSQL 输入数据
	go func() {
		defer wg.Done()
		// 创建定时器，每秒触发一次数据生成
		ticker := time.NewTicker(1 * time.Second)
		defer ticker.Stop()
		for {
			select {
			case <-ticker.C:
				// 每秒生成10条随机测试数据，模拟高频数据流
				// 这种数据密度可以测试 StreamSQL 的实时处理能力
				for i := 0; i < 10; i++ {
					// 构造设备数据，包含设备ID、温度和湿度
					randomData := map[string]interface{}{
						"deviceId":    fmt.Sprintf("device%d", rand.Intn(3)+1), // 随机选择 device1, device2, device3
						"temperature": 20.0 + rand.Float64()*10,                // 温度范围: 20-30度
						"humidity":    50.0 + rand.Float64()*20,                // 湿度范围: 50-70%
					}
					// 将数据添加到流中，触发 StreamSQL 的实时处理
					// Emit 会将数据分发到相应的窗口和聚合器中
					ssql.Emit(randomData)
				}

			case <-ctx.Done():
				// 超时或取消信号，停止数据生成
				return
			}
		}
	}()

	// 步骤6: 设置结果处理管道
	resultChan := make(chan interface{}, 10)
	// 添加计算结果回调函数（Sink）
	// 当窗口触发计算时，结果会通过这个回调函数输出
	ssql.stream.AddSink(func(result []map[string]interface{}) {
		// 非阻塞发送，避免阻塞 sink worker
		select {
		case resultChan <- result:
		default:
			// Channel 已满，忽略（非阻塞发送）
		}
	})

	// 步骤7: 启动结果消费者协程
	// 记录收到的结果数量，用于验证测试效果
	var resultCount int64
	var countMutex sync.Mutex
	var consumerWg sync.WaitGroup
	consumerWg.Add(1)
	go func() {
		defer consumerWg.Done()
		for {
			select {
			case <-resultChan:
				// 每当收到一个窗口的计算结果时，计数器加1
				// 注释掉的代码可以用于调试，打印每个结果的详细信息
				//fmt.Printf("打印结果: [%s] %v\n", time.Now().Format("15:04:05.000"), result)
				countMutex.Lock()
				resultCount++
				countMutex.Unlock()
			case <-ctx.Done():
				// 测试超时，退出消费者 goroutine
				// 不关闭 channel，让主程序自动退出时清理
				return
			}
		}
	}()

	// 步骤8: 等待测试完成
	// 等待数据生产者协程结束（30秒超时或手动取消）
	wg.Wait()

	// 停止流处理，确保所有 goroutine 正确退出
	ssql.Stop()

	// 等待一小段时间，确保所有 sink worker 完成当前任务
	// 这样可以确保所有结果都被发送到 channel
	time.Sleep(100 * time.Millisecond)

	// 取消 context，通知消费者 goroutine 退出
	cancel()

	// 等待消费者 goroutine 完成（处理完 channel 中剩余的数据或收到取消信号）
	consumerWg.Wait()

	// 步骤9: 验证测试结果
	// 预期在30秒内应该收到5个窗口的计算结果（每5秒一个窗口）
	// 这验证了 StreamSQL 的窗口触发机制是否正常工作
	countMutex.Lock()
	finalCount := resultCount
	countMutex.Unlock()
	assert.Equal(t, finalCount, int64(5))
}

func TestStreamsql(t *testing.T) {
	streamsql := New()
	defer streamsql.Stop()
	var rsql = "SELECT device,max(temperature) as max_temp,min(humidity) as min_humidity,window_start() as start,window_end() as end FROM stream group by device,SlidingWindow('2s','1s')"
	err := streamsql.Execute(rsql)
	assert.Nil(t, err)
	strm := streamsql.stream
	// 不使用事件时间，不需要时间戳字段
	testData := []map[string]interface{}{
		{"device": "aa", "temperature": 25.0, "humidity": 60},
		{"device": "aa", "temperature": 30.0, "humidity": 55},
		{"device": "bb", "temperature": 22.0, "humidity": 70},
	}

	for _, data := range testData {
		strm.Emit(data)
	}
	// 捕获结果
	resultChan := make(chan interface{}, 10)
	strm.AddSink(func(result []map[string]interface{}) {
		select {
		case resultChan <- result:
		default:
			// 非阻塞发送，避免阻塞
		}
	})

	// 等待窗口触发
	// 由于使用事件时间，需要等待 watermark 推进（IDLETIMEOUT='2s' 会在2秒后推进）
	time.Sleep(3 * time.Second)

	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()

	var actual interface{}
	select {
	case actual = <-resultChan:
		cancel()
	case <-ctx.Done():
		t.Fatal("Timeout waiting for results")
	}

	expected := []map[string]interface{}{
		{
			"device":       "aa",
			"max_temp":     30.0,
			"min_humidity": 55.0,
		},
		{
			"device":       "bb",
			"max_temp":     22.0,
			"min_humidity": 70.0,
		},
	}

	assert.IsType(t, []map[string]interface{}{}, actual)
	resultSlice, ok := actual.([]map[string]interface{})
	require.True(t, ok)
	assert.Len(t, resultSlice, 2)
	for _, expectedResult := range expected {
		found := false
		for _, resultMap := range resultSlice {
			if resultMap["device"] == expectedResult["device"] {
				assert.InEpsilon(t, expectedResult["max_temp"].(float64), resultMap["max_temp"].(float64), 0.0001)
				assert.InEpsilon(t, expectedResult["min_humidity"].(float64), resultMap["min_humidity"].(float64), 0.0001)
				// 事件时间模式下，窗口时间基于事件时间戳，检查字段存在和有效性
				assert.Contains(t, resultMap, "start")
				assert.Contains(t, resultMap, "end")
				start, ok1 := resultMap["start"].(int64)
				end, ok2 := resultMap["end"].(int64)
				assert.True(t, ok1 && ok2, "start and end should be int64")
				assert.Greater(t, end, start, "end should be greater than start")
				found = true
				break
			}
		}
		assert.True(t, found, fmt.Sprintf("Expected result for device %v not found", expectedResult["device"]))
	}
}

func TestStreamsqlWithoutGroupBy(t *testing.T) {
	streamsql := New()
	defer streamsql.Stop()
	var rsql = "SELECT max(temperature) as max_temp,min(humidity) as min_humidity,window_start() as start,window_end() as end FROM stream SlidingWindow('2s','1s')"
	err := streamsql.Execute(rsql)
	assert.Nil(t, err)
	strm := streamsql.stream
	// 不使用事件时间，不需要时间戳字段
	testData := []map[string]interface{}{
		{"device": "aa", "temperature": 25.0, "humidity": 60},
		{"device": "aa", "temperature": 30.0, "humidity": 55},
		{"device": "bb", "temperature": 22.0, "humidity": 70},
	}

	for _, data := range testData {
		strm.Emit(data)
	}
	// 捕获结果
	resultChan := make(chan interface{}, 10)
	strm.AddSink(func(result []map[string]interface{}) {
		select {
		case resultChan <- result:
		default:
			// 非阻塞发送
		}
	})

	// 等待窗口触发（事件时间模式需要等待 watermark 推进）
	time.Sleep(3 * time.Second)

	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()

	var actual interface{}
	select {
	case actual = <-resultChan:
		cancel()
	case <-ctx.Done():
		t.Fatal("Timeout waiting for results")
	}

	expected := []map[string]interface{}{
		{
			"max_temp":     30.0,
			"min_humidity": 55.0,
		},
	}

	assert.IsType(t, []map[string]interface{}{}, actual)
	resultSlice, ok := actual.([]map[string]interface{})
	require.True(t, ok)
	assert.Len(t, resultSlice, 1)
	for _, expectedResult := range expected {
		for _, resultMap := range resultSlice {
			assert.InEpsilon(t, expectedResult["max_temp"].(float64), resultMap["max_temp"].(float64), 0.0001)
			assert.InEpsilon(t, expectedResult["min_humidity"].(float64), resultMap["min_humidity"].(float64), 0.0001)
			// 事件时间模式下，窗口时间基于事件时间戳，检查字段存在
			assert.Contains(t, resultMap, "start")
			assert.Contains(t, resultMap, "end")
			start, ok1 := resultMap["start"].(int64)
			end, ok2 := resultMap["end"].(int64)
			assert.True(t, ok1 && ok2, "start and end should be int64")
			assert.Greater(t, end, start, "end should be greater than start")
		}
	}
}

func TestStreamsqlDistinct(t *testing.T) {
	streamsql := New()
	defer streamsql.Stop()

	// 测试 SELECT DISTINCT 功能 - 使用聚合函数和 GROUP BY
	var rsql = "SELECT DISTINCT device, AVG(temperature) as avg_temp FROM stream GROUP BY device, TumblingWindow('1s')"
	err := streamsql.Execute(rsql)
	assert.Nil(t, err)
	strm := streamsql.stream

	//fmt.Println("开始测试 SELECT DISTINCT 功能")

	// 添加测试数据，包含重复的设备数据
	// 不使用事件时间，不需要时间戳字段
	testData := []map[string]interface{}{
		{"device": "aa", "temperature": 25.0},
		{"device": "aa", "temperature": 35.0}, // 相同设备，不同温度
		{"device": "bb", "temperature": 22.0},
		{"device": "bb", "temperature": 28.0}, // 相同设备，不同温度
		{"device": "cc", "temperature": 30.0},
	}

	// 添加数据
	//fmt.Println("添加测试数据")
	for _, data := range testData {
		strm.Emit(data)
	}

	// 创建结果接收通道
	resultChan := make(chan interface{}, 10)

	// 添加结果回调
	strm.AddSink(func(result []map[string]interface{}) {
		//fmt.Printf("接收到结果: %v\n", result)
		// 使用 recover 防止在 channel 关闭后发送数据导致 panic
		defer func() {
			if r := recover(); r != nil {
				// Channel 已关闭，忽略错误
			}
		}()
		select {
		case resultChan <- result:
		default:
			// 非阻塞发送
		}
	})

	// 等待窗口触发（处理时间模式）
	//fmt.Println("等待窗口初始化...")
	time.Sleep(1500 * time.Millisecond)

	// 手动触发窗口
	//fmt.Println("手动触发窗口")
	strm.Window.Trigger()

	// 等待结果，增加超时时间以确保窗口有足够时间触发
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	var actual interface{}
	select {
	case actual = <-resultChan:
		//fmt.Println("成功接收到结果")
		cancel()
	case <-ctx.Done():
		t.Fatal("测试超时，未收到结果")
	}

	// 验证结果
	resultSlice, ok := actual.([]map[string]interface{})
	require.True(t, ok, "结果应该是[]map[string]interface{}类型")

	// 验证去重后的结果数量
	assert.Len(t, resultSlice, 3, "应该有3个设备的聚合结果")

	// 检查是否包含所有预期的设备
	deviceFound := make(map[string]bool)
	for _, result := range resultSlice {
		device, ok := result["device"].(string)
		if ok {
			deviceFound[device] = true
		}
	}

	assert.True(t, deviceFound["aa"], "结果应包含设备aa")
	assert.True(t, deviceFound["bb"], "结果应包含设备bb")
	assert.True(t, deviceFound["cc"], "结果应包含设备cc")

	// 验证聚合结果 - aa设备的平均温度应为(25+35)/2=30
	for _, result := range resultSlice {
		device, _ := result["device"].(string)
		avgTemp, ok := result["avg_temp"].(float64)

		assert.True(t, ok, "avg_temp应该是float64类型")

		if device == "aa" {
			assert.InEpsilon(t, 30.0, avgTemp, 0.001, "aa设备的平均温度应为30")
		} else if device == "bb" {
			assert.InEpsilon(t, 25.0, avgTemp, 0.001, "bb设备的平均温度应为25")
		} else if device == "cc" {
			assert.InEpsilon(t, 30.0, avgTemp, 0.001, "cc设备的平均温度应为30")
		}
	}

	//fmt.Println("测试完成")
}

func TestStreamsqlLimit(t *testing.T) {
	// 测试场景1：简单LIMIT功能，不使用窗口函数
	t.Run("简单LIMIT查询", func(t *testing.T) {
		streamsql := New()
		defer streamsql.Stop()

		var rsql = "SELECT * FROM stream LIMIT 2"
		err := streamsql.Execute(rsql)
		assert.Nil(t, err)
		strm := streamsql.stream

		// 创建结果接收通道
		resultChan := make(chan interface{}, 10)

		// 添加结果接收器
		strm.AddSink(func(result []map[string]interface{}) {
			resultChan <- result
		})

		// 添加测试数据
		testData := []map[string]interface{}{
			{"device": "aa", "temperature": 25.0},
			{"device": "bb", "temperature": 22.0},
			{"device": "cc", "temperature": 30.0},
			{"device": "dd", "temperature": 28.0},
		}

		// 实时验证：添加一条数据，立即验证一条结果
		for i, data := range testData {
			// 添加数据
			strm.Emit(data)

			// 立即等待并验证结果
			ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
			select {
			case result := <-resultChan:
				// 验证结果格式
				resultSlice, ok := result.([]map[string]interface{})
				require.True(t, ok, "结果应该是[]map[string]interface{}类型")

				// 验证LIMIT限制：每个batch最多2条记录
				assert.LessOrEqual(t, len(resultSlice), 2, "每个batch最多2条记录")
				assert.Greater(t, len(resultSlice), 0, "应该有结果")

				// 验证字段
				for _, item := range resultSlice {
					assert.Contains(t, item, "device", "结果应包含device字段")
					assert.Contains(t, item, "temperature", "结果应包含temperature字段")
				}
				_ = i
				//t.Logf("第%d条数据处理完成，收到%d条结果记录", i+1, len(resultSlice))
				cancel()
			case <-ctx.Done():
				cancel()
				//t.Fatalf("第%d条数据添加后超时，未收到实时结果", i+1)
			}
		}

		// 验证总体处理：由于LIMIT 2，应该处理完4条数据
		//t.Log("所有数据都得到了实时处理，符合非聚合场景的流处理特性")
	})

	// 测试场景2：聚合查询 + LIMIT
	t.Run("聚合查询与LIMIT", func(t *testing.T) {
		streamsql := New()
		defer streamsql.Stop()

		var rsql = "SELECT device, avg(temperature) as avg_temp, count(*) as cnt FROM stream GROUP BY device LIMIT 2"
		err := streamsql.Execute(rsql)
		assert.Nil(t, err)
		strm := streamsql.stream

		// 创建结果接收通道
		resultChan := make(chan interface{}, 10)

		// 添加结果回调
		strm.AddSink(func(result []map[string]interface{}) {
			resultChan <- result
		})

		// 添加测试数据 - 多个设备的温度数据
		testData := []map[string]interface{}{
			{"device": "sensor1", "temperature": 20.0},
			{"device": "sensor1", "temperature": 22.0},
			{"device": "sensor2", "temperature": 25.0},
			{"device": "sensor2", "temperature": 27.0},
			{"device": "sensor3", "temperature": 30.0},
			{"device": "sensor3", "temperature": 32.0},
			{"device": "sensor4", "temperature": 35.0},
			{"device": "sensor4", "temperature": 37.0},
		}

		// 添加数据
		for _, data := range testData {
			strm.Emit(data)
		}

		// 等待聚合
		time.Sleep(500 * time.Millisecond)

		// 手动触发窗口
		strm.Window.Trigger()

		// 等待结果
		ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
		defer cancel()

		var actual interface{}
		select {
		case actual = <-resultChan:
			cancel()
		case <-ctx.Done():
			t.Fatal("测试超时，未收到聚合结果")
		}

		// 验证聚合结果
		resultSlice, ok := actual.([]map[string]interface{})
		require.True(t, ok, "结果应该是[]map[string]interface{}类型")

		// LIMIT 2 应该只返回2个设备的聚合结果
		assert.LessOrEqual(t, len(resultSlice), 2, "聚合结果应该限制在2条以内")
		assert.Greater(t, len(resultSlice), 0, "应该有聚合结果")

		// 验证聚合字段
		for _, result := range resultSlice {
			assert.Contains(t, result, "device", "结果应包含device字段")
			assert.Contains(t, result, "avg_temp", "结果应包含avg_temp字段")
			assert.Contains(t, result, "cnt", "结果应包含cnt字段")

			// 验证聚合值的类型和合理性
			avgTemp, ok := result["avg_temp"].(float64)
			assert.True(t, ok, "avg_temp应该是float64类型")
			assert.Greater(t, avgTemp, 0.0, "平均温度应该大于0")

			cnt, ok := result["cnt"].(float64)
			assert.True(t, ok, "cnt应该是float64类型")
			assert.GreaterOrEqual(t, cnt, 1.0, "计数应该至少为1")
		}
	})

	// 测试场景3：窗口聚合 + LIMIT
	t.Run("窗口聚合与LIMIT", func(t *testing.T) {
		streamsql := New()
		defer streamsql.Stop()

		var rsql = "SELECT device, max(temperature) as max_temp, min(temperature) as min_temp FROM stream GROUP BY device, TumblingWindow('1s') LIMIT 3"
		err := streamsql.Execute(rsql)
		assert.Nil(t, err)
		strm := streamsql.stream

		// 创建结果接收通道
		resultChan := make(chan interface{}, 10)

		// 添加结果回调
		strm.AddSink(func(result []map[string]interface{}) {
			resultChan <- result
		})

		// 添加测试数据 - 5个设备的数据
		testData := []map[string]interface{}{
			{"device": "dev1", "temperature": 20.0},
			{"device": "dev2", "temperature": 25.0},
			{"device": "dev3", "temperature": 30.0},
			{"device": "dev4", "temperature": 35.0},
			{"device": "dev5", "temperature": 40.0},
		}

		// 添加数据
		for _, data := range testData {
			strm.Emit(data)
		}

		// 等待窗口触发
		time.Sleep(1200 * time.Millisecond) // 等待超过窗口大小

		// 等待结果
		ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
		defer cancel()

		var actual interface{}
		select {
		case actual = <-resultChan:
			cancel()
		case <-ctx.Done():
			t.Fatal("测试超时，未收到窗口聚合结果")
		}

		// 验证窗口聚合结果
		resultSlice, ok := actual.([]map[string]interface{})
		require.True(t, ok, "结果应该是[]map[string]interface{}类型")

		// LIMIT 3 应该只返回3个设备的聚合结果
		assert.LessOrEqual(t, len(resultSlice), 3, "窗口聚合结果应该限制在3条以内")
		assert.Greater(t, len(resultSlice), 0, "应该有窗口聚合结果")

		// 验证聚合字段
		for _, result := range resultSlice {
			assert.Contains(t, result, "device", "结果应包含device字段")
			assert.Contains(t, result, "max_temp", "结果应包含max_temp字段")
			assert.Contains(t, result, "min_temp", "结果应包含min_temp字段")

			// 验证最大值和最小值
			maxTemp, ok := result["max_temp"].(float64)
			assert.True(t, ok, "max_temp应该是float64类型")
			minTemp, ok := result["min_temp"].(float64)
			assert.True(t, ok, "min_temp应该是float64类型")
			assert.GreaterOrEqual(t, maxTemp, minTemp, "最大值应该大于等于最小值")
		}
	})

	// 测试场景4：HAVING + LIMIT 组合
	t.Run("HAVING与LIMIT组合", func(t *testing.T) {
		streamsql := New()
		defer streamsql.Stop()

		var rsql = "SELECT device, avg(temperature) as avg_temp FROM stream GROUP BY device HAVING avg_temp > 25 LIMIT 2"
		err := streamsql.Execute(rsql)
		assert.Nil(t, err)
		strm := streamsql.stream

		// 创建结果接收通道
		resultChan := make(chan interface{}, 10)

		// 添加结果回调
		strm.AddSink(func(result []map[string]interface{}) {
			resultChan <- result
		})

		// 添加测试数据 - 设计一些平均温度大于25的设备
		testData := []map[string]interface{}{
			{"device": "cold_sensor", "temperature": 15.0},
			{"device": "cold_sensor", "temperature": 18.0}, // 平均16.5，不满足条件
			{"device": "warm_sensor1", "temperature": 26.0},
			{"device": "warm_sensor1", "temperature": 28.0}, // 平均27，满足条件
			{"device": "warm_sensor2", "temperature": 30.0},
			{"device": "warm_sensor2", "temperature": 32.0}, // 平均31，满足条件
			{"device": "warm_sensor3", "temperature": 35.0},
			{"device": "warm_sensor3", "temperature": 37.0}, // 平均36，满足条件
		}

		// 添加数据
		for _, data := range testData {
			strm.Emit(data)
		}

		// 等待聚合
		time.Sleep(500 * time.Millisecond)

		// 手动触发窗口
		strm.Window.Trigger()

		// 等待结果
		ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
		defer cancel()

		var actual interface{}
		select {
		case actual = <-resultChan:
			cancel()
		case <-ctx.Done():
			t.Fatal("测试超时，未收到HAVING+LIMIT结果")
		}

		// 验证HAVING + LIMIT结果
		resultSlice, ok := actual.([]map[string]interface{})
		require.True(t, ok, "结果应该是[]map[string]interface{}类型")

		// LIMIT 2 + HAVING条件，最多2条符合条件的结果
		assert.LessOrEqual(t, len(resultSlice), 2, "HAVING+LIMIT结果应该限制在2条以内")

		// 验证所有结果都满足HAVING条件
		for _, result := range resultSlice {
			assert.Contains(t, result, "device", "结果应包含device字段")
			assert.Contains(t, result, "avg_temp", "结果应包含avg_temp字段")

			// 验证不包含cold_sensor（平均温度<25）
			assert.NotEqual(t, "cold_sensor", result["device"], "结果不应包含cold_sensor")

			// 验证平均温度确实大于25
			avgTemp, ok := result["avg_temp"].(float64)
			assert.True(t, ok, "avg_temp应该是float64类型")
			assert.Greater(t, avgTemp, 25.0, "avg_temp应该大于25（满足HAVING条件）")
		}
	})
}

func TestSimpleQuery(t *testing.T) {
	strm := New()
	// 测试结束时确保关闭流处理
	defer strm.Stop()

	// 测试简单查询，不使用窗口函数
	var rsql = "SELECT device, temperature FROM stream"
	err := strm.Execute(rsql)
	assert.Nil(t, err)

	// 创建结果接收通道
	resultChan := make(chan interface{}, 10)

	// 添加sink
	strm.stream.AddSink(func(result []map[string]interface{}) {
		//fmt.Printf("接收到结果: %v\n", result)
		resultChan <- result
	})

	//添加数据
	testData := []map[string]interface{}{
		{"device": "test-device", "temperature": 25.5},
	}

	// 发送数据
	//fmt.Println("添加数据...")
	for _, data := range testData {
		strm.Emit(data)
	}

	// 等待结果
	//fmt.Println("等待结果...")
	//等待结果
	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()
	select {
	case result := <-resultChan:
		//fmt.Printf("收到结果: %v\n", result)
		// 验证结果
		resultSlice, ok := result.([]map[string]interface{})
		require.True(t, ok, "结果应该是[]map[string]interface{}类型")
		require.Len(t, resultSlice, 1, "应该只有一条结果")

		item := resultSlice[0]
		assert.Equal(t, "test-device", item["device"], "device字段应该正确")
		assert.Equal(t, 25.5, item["temperature"], "temperature字段应该正确")
		cancel()
	case <-ctx.Done():
		t.Fatal("测试超时，未收到结果")
	}
	time.Sleep(500 * time.Millisecond)
}

func TestHavingClause(t *testing.T) {
	streamsql := New()
	defer streamsql.Stop()

	// 定义SQL语句，使用HAVING子句
	rsql := "SELECT device, avg(temperature) as avg_temp FROM stream GROUP BY device HAVING avg_temp > 25"
	err := streamsql.Execute(rsql)
	assert.Nil(t, err)
	strm := streamsql.stream

	// 创建结果接收通道
	resultChan := make(chan interface{}, 10)

	// 添加结果回调
	strm.AddSink(func(result []map[string]interface{}) {
		//fmt.Printf("接收到结果: %v\n", result)
		resultChan <- result
	})

	// 添加测试数据，确保有不同的聚合结果
	testData := []map[string]interface{}{
		{"device": "dev1", "temperature": 20.0},
		{"device": "dev1", "temperature": 22.0},
		{"device": "dev2", "temperature": 26.0},
		{"device": "dev2", "temperature": 28.0},
		{"device": "dev3", "temperature": 30.0},
	}

	// 添加数据
	for _, data := range testData {
		strm.Emit(data)
	}

	// 等待窗口初始化
	time.Sleep(500 * time.Millisecond)

	// 手动触发窗口
	strm.Window.Trigger()

	// 等待结果
	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()

	var actual interface{}
	select {
	case actual = <-resultChan:
		cancel()
	case <-ctx.Done():
		t.Fatal("测试超时，未收到结果")
	}

	// 验证结果
	resultSlice, ok := actual.([]map[string]interface{})
	require.True(t, ok, "结果应该是[]map[string]interface{}类型")

	// HAVING avg_temp > 25 应该只返回dev2和dev3
	// 验证结果中不包含dev1
	for _, result := range resultSlice {
		assert.NotEqual(t, "dev1", result["device"], "结果不应包含dev1")
		assert.Contains(t, []string{"dev2", "dev3"}, result["device"], "结果应只包含dev2和dev3")

		// 验证平均温度确实大于25
		avgTemp, ok := result["avg_temp"].(float64)
		assert.True(t, ok, "avg_temp应该是float64类型")
		assert.Greater(t, avgTemp, 25.0, "avg_temp应该大于25")
	}
}

func TestSessionWindow(t *testing.T) {
	streamsql := New()
	defer streamsql.Stop()

	// 使用 SESSION 窗口，超时时间为 2 秒
	rsql := "SELECT device, avg(temperature) as avg_temp FROM stream GROUP BY device, SESSIONWINDOW('2s')"
	err := streamsql.Execute(rsql)
	assert.Nil(t, err)
	strm := streamsql.stream

	// 创建结果接收通道
	resultChan := make(chan interface{}, 10)

	// 添加结果回调
	strm.AddSink(func(result []map[string]interface{}) {
		//fmt.Printf("接收到结果: %v\n", result)
		resultChan <- result
	})

	// 添加测试数据 - 两个设备，不同的时间
	// 不使用事件时间，不需要时间戳字段
	testData := []struct {
		data map[string]interface{}
		wait time.Duration
	}{
		// 第一组数据 - device1
		{map[string]interface{}{"device": "device1", "temperature": 20.0}, 0},
		{map[string]interface{}{"device": "device1", "temperature": 22.0}, 500 * time.Millisecond},

		// 第二组数据 - device2
		{map[string]interface{}{"device": "device2", "temperature": 25.0}, time.Second},
		{map[string]interface{}{"device": "device2", "temperature": 27.0}, 500 * time.Millisecond},

		// 间隔超过会话超时

		// 第三组数据 - device1，新会话
		{map[string]interface{}{"device": "device1", "temperature": 30.0}, 3 * time.Second},
	}

	// 按指定的间隔添加数据
	for _, item := range testData {
		if item.wait > 0 {
			time.Sleep(item.wait)
		}
		strm.Emit(item.data)
	}

	// 等待会话超时，使最后一个会话触发
	time.Sleep(3 * time.Second)

	// 手动触发所有窗口，确保数据被处理
	strm.Window.Trigger()

	// 收集结果
	var results []interface{}
	var resultsMutex sync.Mutex

	// 等待接收结果
	timeout := time.After(5 * time.Second)
	done := false

	for !done {
		select {
		case result := <-resultChan:
			resultsMutex.Lock()
			results = append(results, result)
			resultCount := len(results)
			resultsMutex.Unlock()
			// 我们期望至少 3 个会话结果
			if resultCount >= 3 {
				done = true
			}
		case <-timeout:
			// 超时，可能没有收到足够的结果
			done = true
		}
	}

	// 验证结果
	resultsMutex.Lock()
	resultCount := len(results)
	resultsCopy := make([]interface{}, len(results))
	copy(resultsCopy, results)
	resultsMutex.Unlock()

	assert.GreaterOrEqual(t, resultCount, 2, "应该至少收到两个会话的结果")

	// 检查结果中是否包含两个设备的会话
	hasDevice1 := false
	hasDevice2 := false

	for _, result := range resultsCopy {
		resultSlice, ok := result.([]map[string]interface{})
		assert.True(t, ok, "结果应该是[]map[string]interface{}类型")

		for _, item := range resultSlice {
			device, ok := item["device"].(string)
			assert.True(t, ok, "device字段应该是string类型")

			if device == "device1" {
				hasDevice1 = true
			} else if device == "device2" {
				hasDevice2 = true
			}
		}
	}

	assert.True(t, hasDevice1, "结果中应该包含device1的会话")
	assert.True(t, hasDevice2, "结果中应该包含device2的会话")
}

func TestExpressionInAggregation(t *testing.T) {
	streamsql := New()
	defer streamsql.Stop()

	// 测试在聚合函数中使用表达式
	var rsql = "SELECT device, AVG(temperature * 1.8 + 32) as fahrenheit FROM stream GROUP BY device, TumblingWindow('1s')"
	err := streamsql.Execute(rsql)
	assert.Nil(t, err)
	strm := streamsql.stream

	//fmt.Println("开始测试表达式功能")

	// 添加测试数据，温度使用摄氏度
	// 不使用事件时间，不需要时间戳字段
	testData := []map[string]interface{}{
		{"device": "aa", "temperature": 0.0},   // 华氏度应为 32
		{"device": "aa", "temperature": 100.0}, // 华氏度应为 212
		{"device": "bb", "temperature": 20.0},  // 华氏度应为 68
		{"device": "bb", "temperature": 30.0},  // 华氏度应为 86
	}

	// 添加数据
	//fmt.Println("添加测试数据")
	for _, data := range testData {
		strm.Emit(data)
	}

	// 创建结果接收通道
	resultChan := make(chan interface{}, 10)

	// 添加结果回调
	strm.AddSink(func(result []map[string]interface{}) {
		//fmt.Printf("接收到结果: %v\n", result)
		resultChan <- result
	})

	// 等待窗口触发（处理时间模式）
	//fmt.Println("等待窗口初始化...")
	time.Sleep(1 * time.Second)

	// 手动触发窗口
	//fmt.Println("手动触发窗口")
	strm.Window.Trigger()

	// 等待结果
	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()

	var actual interface{}
	select {
	case actual = <-resultChan:
		//fmt.Println("成功接收到结果")
		cancel()
	case <-ctx.Done():
		t.Fatal("测试超时，未收到结果")
	}

	// 验证结果
	resultSlice, ok := actual.([]map[string]interface{})
	require.True(t, ok, "结果应该是[]map[string]interface{}类型")

	// 验证结果数量
	assert.Len(t, resultSlice, 2, "应该有2个设备的聚合结果")

	// 检查设备及其华氏度温度
	for _, result := range resultSlice {
		device, _ := result["device"].(string)
		fahrenheit, ok := result["fahrenheit"].(float64)

		assert.True(t, ok, "fahrenheit应该是float64类型")

		if device == "aa" {
			// (0 + 100)/2 = 50 摄氏度，转华氏度为 50*1.8+32 = 122
			assert.InEpsilon(t, 122.0, fahrenheit, 0.001, "aa设备的平均华氏温度应为122")
		} else if device == "bb" {
			// (20 + 30)/2 = 25 摄氏度，转华氏度为 25*1.8+32 = 77
			assert.InEpsilon(t, 77.0, fahrenheit, 0.001, "bb设备的平均华氏温度应为77")
		}
	}

	//fmt.Println("表达式测试完成")
}

func TestAdvancedFunctionsInSQL(t *testing.T) {
	streamsql := New()
	defer streamsql.Stop()

	// 测试使用新函数系统的复杂SQL查询
	var rsql = "SELECT device, AVG(abs(temperature - 20)) as abs_diff, CONCAT(device, '_processed') as device_name FROM stream GROUP BY device, TumblingWindow('1s')"
	err := streamsql.Execute(rsql)
	assert.Nil(t, err)
	strm := streamsql.stream

	//fmt.Println("开始测试高级函数功能")

	// 添加测试数据
	// 不使用事件时间，不需要时间戳字段
	testData := []map[string]interface{}{
		{"device": "sensor1", "temperature": 15.0}, // abs(15-20) = 5
		{"device": "sensor1", "temperature": 25.0}, // abs(25-20) = 5
		{"device": "sensor2", "temperature": 18.0}, // abs(18-20) = 2
		{"device": "sensor2", "temperature": 22.0}, // abs(22-20) = 2
	}

	// 添加数据
	//fmt.Println("添加测试数据")
	for _, data := range testData {
		strm.Emit(data)
	}

	// 创建结果接收通道
	resultChan := make(chan interface{}, 10)

	// 添加结果回调
	strm.AddSink(func(result []map[string]interface{}) {
		//fmt.Printf("接收到结果: %v\n", result)
		resultChan <- result
	})

	// 等待窗口触发（处理时间模式）
	//fmt.Println("等待窗口初始化...")
	time.Sleep(1 * time.Second)

	// 手动触发窗口
	//fmt.Println("手动触发窗口")
	strm.Window.Trigger()

	// 等待结果
	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()

	var actual interface{}
	select {
	case actual = <-resultChan:
		//fmt.Println("成功接收到结果")
		cancel()
	case <-ctx.Done():
		t.Fatal("测试超时，未收到结果")
	}

	// 验证结果
	resultSlice, ok := actual.([]map[string]interface{})
	require.True(t, ok, "结果应该是[]map[string]interface{}类型")

	// 验证结果数量
	assert.Len(t, resultSlice, 2, "应该有2个设备的聚合结果")

	// 检查设备及其计算结果
	for _, result := range resultSlice {
		device, _ := result["device"].(string)
		absDiff, ok := result["abs_diff"].(float64)
		deviceName, ok2 := result["device_name"].(string)

		assert.True(t, ok, "abs_diff应该是float64类型")
		assert.True(t, ok2, "device_name应该是string类型")

		if device == "sensor1" {
			// (abs(15-20) + abs(25-20))/2 = (5+5)/2 = 5
			assert.InEpsilon(t, 5.0, absDiff, 0.001, "sensor1的平均绝对差应为5")
			assert.Equal(t, "sensor1_processed", deviceName)
		} else if device == "sensor2" {
			// (abs(18-20) + abs(22-20))/2 = (2+2)/2 = 2
			assert.InEpsilon(t, 2.0, absDiff, 0.001, "sensor2的平均绝对差应为2")
			assert.Equal(t, "sensor2_processed", deviceName)
		}
	}

	//fmt.Println("高级函数测试完成")
}

func TestCustomFunctionInSQL(t *testing.T) {
	// 注册自定义函数：温度华氏度转摄氏度
	err := functions.RegisterCustomFunction("fahrenheit_to_celsius", functions.TypeCustom, "温度转换", "华氏度转摄氏度", 1, 1,
		func(ctx *functions.FunctionContext, args []interface{}) (interface{}, error) {
			fahrenheit := cast.ToFloat64(args[0])
			celsius := (fahrenheit - 32) * 5 / 9
			return celsius, nil
		})
	assert.NoError(t, err)
	defer functions.Unregister("fahrenheit_to_celsius")

	streamsql := New()
	defer streamsql.Stop()

	// 测试使用自定义函数的SQL查询
	var rsql = "SELECT device, AVG(fahrenheit_to_celsius(temperature)) as avg_celsius FROM stream GROUP BY device, TumblingWindow('1s')"
	err = streamsql.Execute(rsql)
	assert.Nil(t, err)
	strm := streamsql.stream

	//fmt.Println("开始测试自定义函数功能")

	// 添加测试数据（华氏度）
	// 不使用事件时间，不需要时间戳字段
	testData := []map[string]interface{}{
		{"device": "thermometer1", "temperature": 32.0},  // 0°C
		{"device": "thermometer1", "temperature": 212.0}, // 100°C
		{"device": "thermometer2", "temperature": 68.0},  // 20°C
		{"device": "thermometer2", "temperature": 86.0},  // 30°C
	}

	// 添加数据
	//fmt.Println("添加测试数据")
	for _, data := range testData {
		strm.Emit(data)
	}

	// 创建结果接收通道
	resultChan := make(chan interface{}, 10)

	// 添加结果回调
	strm.AddSink(func(result []map[string]interface{}) {
		//fmt.Printf("接收到结果: %v\n", result)
		resultChan <- result
	})

	// 等待窗口触发（处理时间模式）
	//fmt.Println("等待窗口初始化...")
	time.Sleep(1 * time.Second)

	// 手动触发窗口
	//fmt.Println("手动触发窗口")
	strm.Window.Trigger()

	// 等待结果
	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()

	var actual interface{}
	select {
	case actual = <-resultChan:
		//fmt.Println("成功接收到结果")
		cancel()
	case <-ctx.Done():
		t.Fatal("测试超时，未收到结果")
	}

	// 验证结果
	resultSlice, ok := actual.([]map[string]interface{})
	require.True(t, ok, "结果应该是[]map[string]interface{}类型")

	// 验证结果数量
	assert.Len(t, resultSlice, 2, "应该有2个设备的聚合结果")

	// 检查设备及其计算结果
	for _, result := range resultSlice {
		device, _ := result["device"].(string)
		avgCelsius, ok := result["avg_celsius"].(float64)

		assert.True(t, ok, "avg_celsius应该是float64类型")

		if device == "thermometer1" {
			// (0 + 100)/2 = 50°C
			assert.InEpsilon(t, 50.0, avgCelsius, 0.001, "thermometer1的平均摄氏温度应为50")
		} else if device == "thermometer2" {
			// (20 + 30)/2 = 25°C
			assert.InEpsilon(t, 25.0, avgCelsius, 0.001, "thermometer2的平均摄氏温度应为25")
		}
	}

	//fmt.Println("自定义函数测试完成")
}

func TestNewAggregateFunctionsInSQL(t *testing.T) {
	streamsql := New()
	defer streamsql.Stop()

	// 测试使用新聚合函数的SQL查询
	var rsql = "SELECT device, collect(temperature) as temp_values, last_value(temperature) as last_temp, merge_agg(status) as all_status FROM stream GROUP BY device, TumblingWindow('1s')"
	err := streamsql.Execute(rsql)
	assert.Nil(t, err)
	strm := streamsql.stream

	//fmt.Println("开始测试新聚合函数功能")

	// 添加测试数据
	// 不使用事件时间，不需要时间戳字段
	testData := []map[string]interface{}{
		{"device": "sensor1", "temperature": 15.0, "status": "good"},
		{"device": "sensor1", "temperature": 25.0, "status": "ok"},
		{"device": "sensor2", "temperature": 18.0, "status": "good"},
		{"device": "sensor2", "temperature": 22.0, "status": "warning"},
	}

	// 添加数据
	//fmt.Println("添加测试数据")
	for _, data := range testData {
		strm.Emit(data)
	}

	// 创建结果接收通道
	resultChan := make(chan interface{}, 10)

	// 添加结果回调
	strm.AddSink(func(result []map[string]interface{}) {
		//fmt.Printf("接收到结果: %v\n", result)
		resultChan <- result
	})

	// 等待窗口触发（处理时间模式）
	//fmt.Println("等待窗口初始化...")
	time.Sleep(1 * time.Second)

	// 手动触发窗口
	//fmt.Println("手动触发窗口")
	strm.Window.Trigger()

	// 等待结果
	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()

	var actual interface{}
	select {
	case actual = <-resultChan:
		//fmt.Println("成功接收到结果")
		cancel()
	case <-ctx.Done():
		t.Fatal("测试超时，未收到结果")
	}

	// 验证结果
	resultSlice, ok := actual.([]map[string]interface{})
	require.True(t, ok, "结果应该是[]map[string]interface{}类型")

	// 验证结果数量
	assert.Len(t, resultSlice, 2, "应该有2个设备的聚合结果")

	// 检查设备及其聚合结果
	for _, result := range resultSlice {
		device, _ := result["device"].(string)
		tempValues, ok1 := result["temp_values"]
		lastTemp, ok2 := result["last_temp"]
		allStatus, ok3 := result["all_status"].(string)

		assert.True(t, ok1, "temp_values应该存在")
		assert.True(t, ok2, "last_temp应该存在")
		assert.True(t, ok3, "all_status应该是string类型")

		if device == "sensor1" {
			// collect函数应该收集[15.0, 25.0]
			values, ok := tempValues.([]interface{})
			assert.True(t, ok, "temp_values应该是数组")
			assert.Len(t, values, 2, "sensor1应该有2个温度值")
			assert.Contains(t, values, 15.0)
			assert.Contains(t, values, 25.0)

			// last_value应该是25.0
			assert.Equal(t, 25.0, lastTemp)

			// merge_agg应该是"good,ok"
			assert.Equal(t, "good,ok", allStatus)
		} else if device == "sensor2" {
			// collect函数应该收集[18.0, 22.0]
			values, ok := tempValues.([]interface{})
			assert.True(t, ok, "temp_values应该是数组")
			assert.Len(t, values, 2, "sensor2应该有2个温度值")
			assert.Contains(t, values, 18.0)
			assert.Contains(t, values, 22.0)

			// last_value应该是22.0
			assert.Equal(t, 22.0, lastTemp)

			// merge_agg应该是"good,warning"
			assert.Equal(t, "good,warning", allStatus)
		}
	}

	//fmt.Println("新聚合函数测试完成")
}

func TestStatisticalAggregateFunctionsInSQL(t *testing.T) {
	streamsql := New()
	defer streamsql.Stop()

	// 测试使用统计聚合函数的SQL查询
	var rsql = "SELECT device, stddevs(temperature) as sample_stddev, var(temperature) as population_var, vars(temperature) as sample_var FROM stream GROUP BY device, TumblingWindow('1s')"
	err := streamsql.Execute(rsql)
	assert.Nil(t, err)
	strm := streamsql.stream

	//fmt.Println("开始测试统计聚合函数功能")

	// 添加测试数据
	// 不使用事件时间，不需要时间戳字段
	testData := []map[string]interface{}{
		{"device": "sensor1", "temperature": 10.0},
		{"device": "sensor1", "temperature": 20.0},
		{"device": "sensor1", "temperature": 30.0},
		{"device": "sensor2", "temperature": 15.0},
		{"device": "sensor2", "temperature": 25.0},
	}

	// 添加数据
	//fmt.Println("添加测试数据")
	for _, data := range testData {
		strm.Emit(data)
	}

	// 创建结果接收通道
	resultChan := make(chan interface{}, 10)

	// 添加结果回调
	strm.AddSink(func(result []map[string]interface{}) {
		//fmt.Printf("接收到结果: %v\n", result)
		resultChan <- result
	})

	// 等待窗口触发（处理时间模式）
	//fmt.Println("等待窗口初始化...")
	time.Sleep(1 * time.Second)

	// 手动触发窗口
	//fmt.Println("手动触发窗口")
	strm.Window.Trigger()

	// 等待结果
	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()

	var actual interface{}
	select {
	case actual = <-resultChan:
		//fmt.Println("成功接收到结果")
		cancel()
	case <-ctx.Done():
		t.Fatal("测试超时，未收到结果")
	}

	// 验证结果
	resultSlice, ok := actual.([]map[string]interface{})
	require.True(t, ok, "结果应该是[]map[string]interface{}类型")

	// 验证结果数量
	assert.Len(t, resultSlice, 2, "应该有2个设备的聚合结果")

	// 检查设备及其统计结果
	for _, result := range resultSlice {
		device, _ := result["device"].(string)
		sampleStddev, ok1 := result["sample_stddev"].(float64)
		populationVar, ok2 := result["population_var"].(float64)
		sampleVar, ok3 := result["sample_var"].(float64)

		assert.True(t, ok1, "sample_stddev应该是float64类型")
		assert.True(t, ok2, "population_var应该是float64类型")
		assert.True(t, ok3, "sample_var应该是float64类型")

		if device == "sensor1" {
			// sensor1: [10, 20, 30], 平均值=20
			// 总体方差 = ((10-20)² + (20-20)² + (30-20)²) / 3 = (100 + 0 + 100) / 3 = 66.67
			// 样本方差 = 200 / 2 = 100
			// 样本标准差 = sqrt(100) = 10
			assert.InEpsilon(t, 10.0, sampleStddev, 0.001, "sensor1的样本标准差应约为10")
			assert.InEpsilon(t, 66.67, populationVar, 0.1, "sensor1的总体方差应约为66.67")
			assert.InEpsilon(t, 100.0, sampleVar, 0.001, "sensor1的样本方差应约为100")
		} else if device == "sensor2" {
			// sensor2: [15, 25], 平均值=20
			// 总体方差 = ((15-20)² + (25-20)²) / 2 = (25 + 25) / 2 = 25
			// 样本方差 = 50 / 1 = 50
			// 样本标准差 = sqrt(50) = 7.07
			assert.InEpsilon(t, 7.07, sampleStddev, 0.1, "sensor2的样本标准差应约为7.07")
			assert.InEpsilon(t, 25.0, populationVar, 0.001, "sensor2的总体方差应约为25")
			assert.InEpsilon(t, 50.0, sampleVar, 0.001, "sensor2的样本方差应约为50")
		}
	}

	//fmt.Println("统计聚合函数测试完成")
}

func TestDeduplicateAggregateInSQL(t *testing.T) {
	streamsql := New()
	defer streamsql.Stop()

	// 测试使用去重聚合函数的SQL查询
	var rsql = "SELECT device, deduplicate(status) as unique_status FROM stream GROUP BY device, TumblingWindow('1s')"
	err := streamsql.Execute(rsql)
	assert.Nil(t, err)
	strm := streamsql.stream

	//fmt.Println("开始测试去重聚合函数功能")

	// 添加测试数据，包含重复的状态
	// 不使用事件时间，不需要时间戳字段
	testData := []map[string]interface{}{
		{"device": "sensor1", "status": "good"},
		{"device": "sensor1", "status": "good"}, // 重复
		{"device": "sensor1", "status": "warning"},
		{"device": "sensor1", "status": "good"}, // 重复
		{"device": "sensor2", "status": "error"},
		{"device": "sensor2", "status": "error"}, // 重复
		{"device": "sensor2", "status": "ok"},
	}

	// 添加数据
	//fmt.Println("添加测试数据")
	for _, data := range testData {
		strm.Emit(data)
	}

	// 创建结果接收通道
	resultChan := make(chan interface{}, 10)

	// 添加结果回调
	strm.AddSink(func(result []map[string]interface{}) {
		//fmt.Printf("接收到结果: %v\n", result)
		resultChan <- result
	})

	// 等待窗口触发（处理时间模式）
	//fmt.Println("等待窗口初始化...")
	time.Sleep(1 * time.Second)

	// 手动触发窗口
	//fmt.Println("手动触发窗口")
	strm.Window.Trigger()

	// 等待结果
	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()

	var actual interface{}
	select {
	case actual = <-resultChan:
		//fmt.Println("成功接收到结果")
		cancel()
	case <-ctx.Done():
		t.Fatal("测试超时，未收到结果")
	}

	// 验证结果
	resultSlice, ok := actual.([]map[string]interface{})
	require.True(t, ok, "结果应该是[]map[string]interface{}类型")

	// 验证结果数量
	assert.Len(t, resultSlice, 2, "应该有2个设备的聚合结果")

	// 检查设备及其去重结果
	for _, result := range resultSlice {
		device, _ := result["device"].(string)
		uniqueStatus, ok := result["unique_status"]

		assert.True(t, ok, "unique_status应该存在")

		if device == "sensor1" {
			// sensor1应该有去重后的状态：["good", "warning"]
			statusArray, ok := uniqueStatus.([]interface{})
			assert.True(t, ok, "unique_status应该是数组")
			assert.Len(t, statusArray, 2, "sensor1应该有2个不同的状态")
			assert.Contains(t, statusArray, "good")
			assert.Contains(t, statusArray, "warning")
		} else if device == "sensor2" {
			// sensor2应该有去重后的状态：["error", "ok"]
			statusArray, ok := uniqueStatus.([]interface{})
			assert.True(t, ok, "unique_status应该是数组")
			assert.Len(t, statusArray, 2, "sensor2应该有2个不同的状态")
			assert.Contains(t, statusArray, "error")
			assert.Contains(t, statusArray, "ok")
		}
	}

	//fmt.Println("去重聚合函数测试完成")
}

func TestExprAggregationFunctions(t *testing.T) {
	streamsql := New()
	defer streamsql.Stop()

	// 测试使用表达式运算的聚合函数SQL查询
	var rsql = `SELECT 
		device,
		avg(temperature * 1.8 + 32) as avg_fahrenheit,  
		stddevs((temperature - 20) * 2) as temp_stddev, 
		var(temperature / 10) as temp_var,             
		collect(temperature + humidity) as temp_hum_sum, 
		last_value(temperature * humidity) as last_temp_hum, 
		merge_agg(device + '_' + status) as device_status,  
		deduplicate(status + '_' + device) as unique_status_device
	FROM stream 
	GROUP BY device, TumblingWindow('1s')`

	err := streamsql.Execute(rsql)
	assert.Nil(t, err)
	strm := streamsql.stream

	//fmt.Println("开始测试表达式聚合函数功能")

	// 添加测试数据
	// 不使用事件时间，不需要时间戳字段
	testData := []map[string]interface{}{
		// device1的数据
		{"device": "device1", "temperature": 20.0, "humidity": 60.0, "status": "normal"},  // 华氏度=68, 偏差=0, 和=80
		{"device": "device1", "temperature": 25.0, "humidity": 65.0, "status": "warning"}, // 华氏度=77, 偏差=10, 和=90
		{"device": "device1", "temperature": 30.0, "humidity": 70.0, "status": "normal"},  // 华氏度=86, 偏差=20, 和=100

		// device2的数据
		{"device": "device2", "temperature": 15.0, "humidity": 55.0, "status": "error"},  // 华氏度=59, 偏差=-10, 和=70
		{"device": "device2", "temperature": 18.0, "humidity": 58.0, "status": "normal"}, // 华氏度=64.4, 偏差=-4, 和=76
		{"device": "device2", "temperature": 22.0, "humidity": 62.0, "status": "error"},  // 华氏度=71.6, 偏差=4, 和=84
	}

	// 添加数据
	//fmt.Println("添加测试数据")
	for _, data := range testData {
		strm.Emit(data)
	}

	// 创建结果接收通道
	resultChan := make(chan interface{}, 10)

	// 添加结果回调
	strm.AddSink(func(result []map[string]interface{}) {
		//fmt.Printf("接收到结果: %v\n", result)
		resultChan <- result
	})

	// 等待窗口触发（处理时间模式）
	//fmt.Println("等待窗口初始化...")
	time.Sleep(1 * time.Second)

	// 手动触发窗口
	//fmt.Println("手动触发窗口")
	strm.Window.Trigger()

	// 等待结果
	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()

	var actual interface{}
	select {
	case actual = <-resultChan:
		//fmt.Println("成功接收到结果")
		cancel()
	case <-ctx.Done():
		t.Fatal("测试超时，未收到结果")
	}

	// 验证结果
	resultSlice, ok := actual.([]map[string]interface{})
	require.True(t, ok, "结果应该是[]map[string]interface{}类型")

	// 验证结果数量
	assert.Len(t, resultSlice, 2, "应该有2个设备的聚合结果")

	// 检查设备及其计算结果
	for _, result := range resultSlice {
		device, _ := result["device"].(string)
		avgFahrenheit, ok1 := result["avg_fahrenheit"].(float64)
		tempStddev, ok2 := result["temp_stddev"].(float64)
		tempVar, ok3 := result["temp_var"].(float64)
		tempHumSum, ok4 := result["temp_hum_sum"]
		lastTempHum, ok5 := result["last_temp_hum"].(float64)
		deviceStatus, ok6 := result["device_status"].(string)
		uniqueStatusDevice, ok7 := result["unique_status_device"]

		assert.True(t, ok1, "avg_fahrenheit应该是float64类型")
		assert.True(t, ok2, "temp_stddev应该是float64类型")
		assert.True(t, ok3, "temp_var应该是float64类型")
		assert.True(t, ok4, "temp_hum_sum应该存在")
		assert.True(t, ok5, "last_temp_hum应该是float64类型")
		assert.True(t, ok6, "device_status应该是string类型")
		assert.True(t, ok7, "unique_status_device应该存在")

		if device == "device1" {
			// device1的验证
			// 平均华氏度: (68 + 77 + 86) / 3 = 77
			assert.InEpsilon(t, 77.0, avgFahrenheit, 0.1, "device1的平均华氏度应约为77")

			// 温度偏差标准差: sqrt(((0-10)² + (10-10)² + (20-10)²) / 2) = sqrt(200/2) = 10
			assert.InEpsilon(t, 10.0, tempStddev, 0.1, "device1的温度偏差标准差应约为10")

			// 温度除以10的方差: ((2-2.5)² + (2.5-2.5)² + (3-2.5)²) / 3 = 0.167
			assert.InEpsilon(t, 0.167, tempVar, 0.01, "device1的温度方差应约为0.167")

			// 温度和湿度的和数组
			tempHumSumArray, ok := tempHumSum.([]interface{})
			assert.True(t, ok, "temp_hum_sum应该是数组")
			assert.Len(t, tempHumSumArray, 3, "device1应该有3个温度和湿度的和")
			assert.Contains(t, tempHumSumArray, 80.0)
			assert.Contains(t, tempHumSumArray, 90.0)
			assert.Contains(t, tempHumSumArray, 100.0)

			// 最后一个温度和湿度的乘积: 30 * 70 = 2100
			assert.InEpsilon(t, 2100.0, lastTempHum, 0.1, "device1的最后一个温度和湿度乘积应为2100")

			// 设备状态组合
			assert.Contains(t, deviceStatus, "device1_normal")
			assert.Contains(t, deviceStatus, "device1_warning")

			// 状态设备组合去重
			uniqueArray, ok := uniqueStatusDevice.([]interface{})
			assert.True(t, ok, "unique_status_device应该是数组")
			assert.Len(t, uniqueArray, 2, "device1应该有2个不同的状态设备组合")
			assert.Contains(t, uniqueArray, "normal_device1")
			assert.Contains(t, uniqueArray, "warning_device1")

		} else if device == "device2" {
			// device2的验证
			// 平均华氏度: (59 + 64.4 + 71.6) / 3 = 65
			assert.InEpsilon(t, 65.0, avgFahrenheit, 0.1, "device2的平均华氏度应约为65")

			// 温度偏差标准差: sqrt(((-10-(-3.33))² + (-4-(-3.33))² + (4-(-3.33))²) / 2) = sqrt(147.33/2) = 7.023
			assert.InEpsilon(t, 7.023, tempStddev, 0.1, "device2的温度偏差标准差应约为7.023")

			// 温度除以10的方差: ((1.5-1.83)² + (1.8-1.83)² + (2.2-1.83)²) / 3 = 0.082
			assert.InEpsilon(t, 0.082, tempVar, 0.01, "device2的温度方差应约为0.082")

			// 温度和湿度的和数组
			tempHumSumArray, ok := tempHumSum.([]interface{})
			assert.True(t, ok, "temp_hum_sum应该是数组")
			assert.Len(t, tempHumSumArray, 3, "device2应该有3个温度和湿度的和")
			assert.Contains(t, tempHumSumArray, 70.0)
			assert.Contains(t, tempHumSumArray, 76.0)
			assert.Contains(t, tempHumSumArray, 84.0)

			// 最后一个温度和湿度的乘积: 22 * 62 = 1364
			assert.InEpsilon(t, 1364.0, lastTempHum, 0.1, "device2的最后一个温度和湿度乘积应为1364")

			// 设备状态组合
			assert.Contains(t, deviceStatus, "device2_error")
			assert.Contains(t, deviceStatus, "device2_normal")

			// 状态设备组合去重
			uniqueArray, ok := uniqueStatusDevice.([]interface{})
			assert.True(t, ok, "unique_status_device应该是数组")
			assert.Len(t, uniqueArray, 2, "device2应该有2个不同的状态设备组合")
			assert.Contains(t, uniqueArray, "error_device2")
			assert.Contains(t, uniqueArray, "normal_device2")
		}
	}

	//fmt.Println("表达式聚合函数测试完成")
}

func TestAnalyticalFunctionsInSQL(t *testing.T) {
	streamsql := New()
	defer streamsql.Stop()

	// 测试使用分析函数的SQL查询
	var rsql = "SELECT device, lag(temperature) as prev_temp, latest(temperature) as current_temp, had_changed(temperature) as temp_changed FROM stream GROUP BY device, TumblingWindow('1s')"
	err := streamsql.Execute(rsql)
	assert.Nil(t, err)
	strm := streamsql.stream

	//fmt.Println("开始测试分析函数功能")

	// 添加测试数据
	// 不使用事件时间，不需要时间戳字段
	testData := []map[string]interface{}{
		{"device": "sensor1", "temperature": 20.0},
		{"device": "sensor1", "temperature": 25.0},
		{"device": "sensor1", "temperature": 25.0}, // 重复值，测试had_changed
		{"device": "sensor2", "temperature": 18.0},
		{"device": "sensor2", "temperature": 22.0},
	}

	// 添加数据
	//fmt.Println("添加测试数据")
	for _, data := range testData {
		strm.Emit(data)
	}

	// 创建结果接收通道
	resultChan := make(chan interface{}, 10)

	// 添加结果回调
	strm.AddSink(func(result []map[string]interface{}) {
		//fmt.Printf("接收到结果: %v\n", result)
		resultChan <- result
	})

	// 等待窗口触发（处理时间模式）
	//fmt.Println("等待窗口初始化...")
	time.Sleep(1 * time.Second)

	// 手动触发窗口
	//fmt.Println("手动触发窗口")
	strm.Window.Trigger()

	// 等待结果
	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()

	var actual interface{}
	select {
	case actual = <-resultChan:
		//fmt.Println("成功接收到结果")
		cancel()
	case <-ctx.Done():
		t.Fatal("测试超时，未收到结果")
	}

	// 验证结果
	resultSlice, ok := actual.([]map[string]interface{})
	require.True(t, ok, "结果应该是[]map[string]interface{}类型")

	// 验证结果数量
	assert.Len(t, resultSlice, 2, "应该有2个设备的聚合结果")

	// 检查设备及其分析函数结果
	for _, result := range resultSlice {
		device, _ := result["device"].(string)

		assert.Contains(t, result, "prev_temp", "结果应包含prev_temp字段")
		assert.Contains(t, result, "current_temp", "结果应包含current_temp字段")
		assert.Contains(t, result, "temp_changed", "结果应包含temp_changed字段")

		if device == "sensor1" {
			// sensor1有3个温度值: 20.0, 25.0, 25.0
			// latest应该返回最新值
			currentTemp := result["current_temp"]
			assert.NotNil(t, currentTemp, "current_temp不应为空")

			// had_changed应该有变化记录
			tempChanged := result["temp_changed"]
			assert.NotNil(t, tempChanged, "temp_changed不应为空")
		} else if device == "sensor2" {
			// sensor2有2个温度值: 18.0, 22.0
			currentTemp := result["current_temp"]
			assert.NotNil(t, currentTemp, "current_temp不应为空")

			tempChanged := result["temp_changed"]
			assert.NotNil(t, tempChanged, "temp_changed不应为空")
		}
	}

	//fmt.Println("分析函数测试完成")
}

func TestLagFunctionInSQL(t *testing.T) {
	streamsql := New()
	defer streamsql.Stop()

	// 测试LAG函数的SQL查询
	var rsql = "SELECT device, lag(temperature) as prev_temp FROM stream GROUP BY device, TumblingWindow('1s')"
	err := streamsql.Execute(rsql)
	assert.Nil(t, err)
	strm := streamsql.stream

	//fmt.Println("开始测试LAG函数功能")

	// 添加测试数据 - 按顺序添加，测试LAG功能
	// 不使用事件时间，不需要时间戳字段
	testData := []map[string]interface{}{
		{"device": "temp_sensor", "temperature": 10.0},
		{"device": "temp_sensor", "temperature": 15.0},
		{"device": "temp_sensor", "temperature": 20.0},
		{"device": "temp_sensor", "temperature": 25.0}, // 最后一个值
	}

	// 添加数据
	//fmt.Println("添加测试数据：", testData)
	for _, data := range testData {
		//fmt.Printf("添加第%d个数据: temperature=%.1f\n", i+1, data.(map[string]interface{})["temperature"])
		strm.Emit(data)
		time.Sleep(100 * time.Millisecond) // 稍微延迟确保顺序
	}

	// 创建结果接收通道
	resultChan := make(chan interface{}, 10)

	// 添加结果回调
	strm.AddSink(func(result []map[string]interface{}) {
		//fmt.Printf("接收到结果: %v\n", result)
		resultChan <- result
	})

	// 等待窗口触发（处理时间模式）
	//fmt.Println("等待窗口初始化...")
	time.Sleep(1 * time.Second)

	// 手动触发窗口
	//fmt.Println("手动触发窗口")
	strm.Window.Trigger()

	// 等待结果
	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()

	var actual interface{}
	select {
	case actual = <-resultChan:
		//fmt.Println("成功接收到结果")
		cancel()
	case <-ctx.Done():
		t.Fatal("测试超时，未收到结果")
	}

	// 验证结果
	resultSlice, ok := actual.([]map[string]interface{})
	require.True(t, ok, "结果应该是[]map[string]interface{}类型")

	// 验证结果数量
	assert.Len(t, resultSlice, 1, "应该有1个设备的聚合结果")

	result := resultSlice[0]
	device, _ := result["device"].(string)
	assert.Equal(t, "temp_sensor", device, "设备名应该正确")

	// 验证字段存在
	assert.Contains(t, result, "prev_temp", "结果应包含prev_temp字段")

	// LAG函数应该返回最后一个值(25.0)的前一个值(20.0)
	// 数据序列：10.0 -> 15.0 -> 20.0 -> 25.0
	// LAG执行过程：
	// - 10.0: 无前值，返回nil
	// - 15.0: 前值10.0，返回10.0
	// - 20.0: 前值15.0，返回15.0
	// - 25.0: 前值20.0，返回20.0 ← 最终结果
	prevTemp := result["prev_temp"]
	//fmt.Printf("LAG函数返回值: %v (期望: 20.0，表示最后值25.0的前一个值)\n", prevTemp)

	// 验证LAG函数返回正确的前一个值
	expectedPrevTemp := 20.0
	if prevTemp != nil {
		prevTempFloat, ok := prevTemp.(float64)
		assert.True(t, ok, "prev_temp应该是float64类型")
		assert.Equal(t, expectedPrevTemp, prevTempFloat, "LAG函数应该返回最后一个值的前一个值(20.0)")
	} else {
		t.Errorf("LAG函数不应该返回nil，期望值: %.1f", expectedPrevTemp)
	}

	//fmt.Println("LAG函数测试完成")
}

func TestHadChangedFunctionInSQL(t *testing.T) {
	streamsql := New()
	defer streamsql.Stop()

	// 测试had_changed函数的SQL查询
	var rsql = "SELECT device, had_changed(temperature) as temp_changed FROM stream GROUP BY device, TumblingWindow('1s')"
	err := streamsql.Execute(rsql)
	assert.Nil(t, err)
	strm := streamsql.stream

	//fmt.Println("开始测试had_changed函数功能")

	// 添加测试数据 - 包含重复值和变化值
	// 不使用事件时间，不需要时间戳字段
	testData := []map[string]interface{}{
		{"device": "monitor", "temperature": 20.0},
		{"device": "monitor", "temperature": 20.0}, // 相同值
		{"device": "monitor", "temperature": 25.0}, // 变化值
		{"device": "monitor", "temperature": 25.0}, // 相同值
		{"device": "monitor", "temperature": 30.0}, // 变化值
	}

	// 添加数据
	//fmt.Println("添加测试数据")
	for _, data := range testData {
		strm.Emit(data)
	}

	// 创建结果接收通道
	resultChan := make(chan interface{}, 10)

	// 添加结果回调
	strm.AddSink(func(result []map[string]interface{}) {
		//fmt.Printf("接收到结果: %v\n", result)
		resultChan <- result
	})

	// 等待窗口触发（处理时间模式）
	//fmt.Println("等待窗口初始化...")
	time.Sleep(1 * time.Second)

	// 手动触发窗口
	//fmt.Println("手动触发窗口")
	strm.Window.Trigger()

	// 等待结果
	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()

	var actual interface{}
	select {
	case actual = <-resultChan:
		//fmt.Println("成功接收到结果")
		cancel()
	case <-ctx.Done():
		t.Fatal("测试超时，未收到结果")
	}

	// 验证结果
	resultSlice, ok := actual.([]map[string]interface{})
	require.True(t, ok, "结果应该是[]map[string]interface{}类型")

	// 验证结果数量
	assert.Len(t, resultSlice, 1, "应该有1个设备的聚合结果")

	result := resultSlice[0]
	device, _ := result["device"].(string)
	assert.Equal(t, "monitor", device, "设备名应该正确")

	// 验证字段存在
	assert.Contains(t, result, "temp_changed", "结果应包含temp_changed字段")

	// had_changed函数应该返回布尔值
	//tempChanged := result["temp_changed"]
	//fmt.Printf("had_changed函数返回值: %v\n", tempChanged)

	//fmt.Println("had_changed函数测试完成")
}

func TestLatestFunctionInSQL(t *testing.T) {
	streamsql := New()
	defer streamsql.Stop()

	// 测试latest函数的SQL查询
	var rsql = "SELECT device, latest(temperature) as current_temp FROM stream GROUP BY device, TumblingWindow('1s')"
	err := streamsql.Execute(rsql)
	assert.Nil(t, err)
	strm := streamsql.stream

	//fmt.Println("开始测试latest函数功能")

	// 添加测试数据
	// 不使用事件时间，不需要时间戳字段
	testData := []map[string]interface{}{
		{"device": "thermometer", "temperature": 10.0},
		{"device": "thermometer", "temperature": 15.0},
		{"device": "thermometer", "temperature": 20.0},
		{"device": "thermometer", "temperature": 25.0}, // 最新值
	}

	// 添加数据
	//fmt.Println("添加测试数据")
	for _, data := range testData {
		strm.Emit(data)
	}

	// 创建结果接收通道
	resultChan := make(chan interface{}, 10)

	// 添加结果回调
	strm.AddSink(func(result []map[string]interface{}) {
		//fmt.Printf("接收到结果: %v\n", result)
		resultChan <- result
	})

	// 等待窗口触发（处理时间模式）
	//fmt.Println("等待窗口初始化...")
	time.Sleep(1 * time.Second)

	// 手动触发窗口
	//fmt.Println("手动触发窗口")
	strm.Window.Trigger()

	// 等待结果
	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()

	var actual interface{}
	select {
	case actual = <-resultChan:
		//fmt.Println("成功接收到结果")
		cancel()
	case <-ctx.Done():
		t.Fatal("测试超时，未收到结果")
	}

	// 验证结果
	resultSlice, ok := actual.([]map[string]interface{})
	require.True(t, ok, "结果应该是[]map[string]interface{}类型")

	// 验证结果数量
	assert.Len(t, resultSlice, 1, "应该有1个设备的聚合结果")

	result := resultSlice[0]
	device, _ := result["device"].(string)
	assert.Equal(t, "thermometer", device, "设备名应该正确")

	// 验证字段存在
	assert.Contains(t, result, "current_temp", "结果应包含current_temp字段")

	// latest函数应该返回最新值25.0
	currentTemp, ok := result["current_temp"].(float64)
	assert.True(t, ok, "current_temp应该是float64类型")
	assert.Equal(t, 25.0, currentTemp, "latest函数应该返回最新值25.0")

	//fmt.Println("latest函数测试完成")
}

func TestChangedColFunctionInSQL(t *testing.T) {
	streamsql := New()
	defer streamsql.Stop()

	// 测试changed_col函数的SQL查询
	var rsql = "SELECT device, changed_col(data) as changed_fields FROM stream GROUP BY device, TumblingWindow('1s')"
	err := streamsql.Execute(rsql)
	assert.Nil(t, err)
	strm := streamsql.stream

	//fmt.Println("开始测试changed_col函数功能")

	// 添加测试数据 - 使用map作为数据测试changed_col
	// 不使用事件时间，不需要时间戳字段
	testData := []map[string]interface{}{
		{
			"device": "datacollector",
			"data":   map[string]interface{}{"temp": 20.0, "humidity": 60.0},
		},
		{
			"device": "datacollector",
			"data":   map[string]interface{}{"temp": 25.0, "humidity": 60.0}, // temp变化
		},
		{
			"device": "datacollector",
			"data":   map[string]interface{}{"temp": 25.0, "humidity": 65.0}, // humidity变化
		},
	}

	// 添加数据
	//fmt.Println("添加测试数据")
	for _, data := range testData {
		strm.Emit(data)
	}

	// 创建结果接收通道
	resultChan := make(chan interface{}, 10)

	// 添加结果回调
	strm.AddSink(func(result []map[string]interface{}) {
		//fmt.Printf("接收到结果: %v\n", result)
		resultChan <- result
	})

	// 等待窗口触发（处理时间模式）
	//fmt.Println("等待窗口初始化...")
	time.Sleep(1 * time.Second)

	// 手动触发窗口
	//fmt.Println("手动触发窗口")
	strm.Window.Trigger()

	// 等待结果
	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()

	var actual interface{}
	select {
	case actual = <-resultChan:
		//fmt.Println("成功接收到结果")
		cancel()
	case <-ctx.Done():
		t.Fatal("测试超时，未收到结果")
	}

	// 验证结果
	resultSlice, ok := actual.([]map[string]interface{})
	require.True(t, ok, "结果应该是[]map[string]interface{}类型")

	// 验证结果数量
	assert.Len(t, resultSlice, 1, "应该有1个设备的聚合结果")

	result := resultSlice[0]
	device, _ := result["device"].(string)
	assert.Equal(t, "datacollector", device, "设备名应该正确")

	// 验证字段存在
	assert.Contains(t, result, "changed_fields", "结果应包含changed_fields字段")

	// changed_col函数应该返回变化的字段列表
	//changedFields := result["changed_fields"]
	//fmt.Printf("changed_col函数返回值: %v\n", changedFields)

	//fmt.Println("changed_col函数测试完成")
}

func TestAnalyticalFunctionsIncrementalComputation(t *testing.T) {
	streamsql := New()
	defer streamsql.Stop()

	// 测试使用分析函数的SQL查询（现在支持增量计算）
	var rsql = "SELECT device, lag(temperature, 1) as prev_temp, latest(temperature) as current_temp, had_changed(status) as status_changed FROM stream GROUP BY device, TumblingWindow('1s')"
	err := streamsql.Execute(rsql)
	assert.Nil(t, err)
	strm := streamsql.stream

	//fmt.Println("开始测试分析函数增量计算功能")

	// 添加测试数据
	// 不使用事件时间，不需要时间戳字段
	testData := []map[string]interface{}{
		{"device": "sensor1", "temperature": 15.0, "status": "good"},
		{"device": "sensor1", "temperature": 25.0, "status": "good"},
		{"device": "sensor1", "temperature": 35.0, "status": "warning"},
		{"device": "sensor2", "temperature": 18.0, "status": "good"},
		{"device": "sensor2", "temperature": 22.0, "status": "ok"},
	}

	// 添加数据
	//fmt.Println("添加测试数据")
	for _, data := range testData {
		strm.Emit(data)
	}

	// 创建结果接收通道
	resultChan := make(chan interface{}, 10)

	// 添加结果回调
	strm.AddSink(func(result []map[string]interface{}) {
		//fmt.Printf("接收到结果: %v\n", result)
		resultChan <- result
	})

	// 等待窗口触发（处理时间模式）
	//fmt.Println("等待窗口初始化...")
	time.Sleep(1 * time.Second)

	// 手动触发窗口
	//fmt.Println("手动触发窗口")
	strm.Window.Trigger()

	// 等待结果
	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()

	var actual interface{}
	select {
	case actual = <-resultChan:
		//fmt.Println("成功接收到结果")
		cancel()
	case <-ctx.Done():
		t.Fatal("测试超时，未收到结果")
	}

	// 验证结果
	resultSlice, ok := actual.([]map[string]interface{})
	require.True(t, ok, "结果应该是[]map[string]interface{}类型")

	// 验证结果数量
	assert.Len(t, resultSlice, 2, "应该有2个设备的聚合结果")

	// 检查设备及其分析结果
	for _, result := range resultSlice {
		device, _ := result["device"].(string)
		currentTemp := result["current_temp"]
		statusChanged := result["status_changed"]

		//fmt.Printf("设备 %s: current_temp=%v, status_changed=%v\n", device, currentTemp, statusChanged)

		if device == "sensor1" {
			// latest函数应该返回最新的温度值35.0
			if currentTemp != nil {
				assert.Equal(t, 35.0, currentTemp)
			}
			// had_changed函数应该检测到状态变化（good -> warning）
			if statusChanged != nil {
				assert.True(t, statusChanged.(bool), "sensor1的状态应该发生了变化")
			}
		} else if device == "sensor2" {
			// latest函数应该返回最新的温度值22.0
			if currentTemp != nil {
				assert.Equal(t, 22.0, currentTemp)
			}
			// had_changed函数应该检测到状态变化（good -> ok）
			if statusChanged != nil {
				assert.True(t, statusChanged.(bool), "sensor2的状态应该发生了变化")
			}
		}
	}

	//fmt.Println("分析函数增量计算测试完成")
}

func TestIncrementalComputationBasic(t *testing.T) {
	streamsql := New()
	defer streamsql.Stop()

	// 测试基本的增量计算聚合函数
	var rsql = "SELECT device, sum(temperature) as total, avg(temperature) as average, count(*) as cnt FROM stream GROUP BY device, TumblingWindow('1s')"
	err := streamsql.Execute(rsql)
	assert.Nil(t, err)
	strm := streamsql.stream

	//fmt.Println("开始测试基本增量计算功能")

	// 添加测试数据
	// 不使用事件时间，不需要时间戳字段
	testData := []map[string]interface{}{
		{"device": "sensor1", "temperature": 10.0},
		{"device": "sensor1", "temperature": 20.0},
		{"device": "sensor1", "temperature": 30.0},
		{"device": "sensor2", "temperature": 15.0},
		{"device": "sensor2", "temperature": 25.0},
	}

	// 添加数据
	//fmt.Println("添加测试数据")
	for _, data := range testData {
		strm.Emit(data)
	}

	// 创建结果接收通道
	resultChan := make(chan interface{}, 10)

	// 添加结果回调
	strm.AddSink(func(result []map[string]interface{}) {
		//fmt.Printf("接收到结果: %v\n", result)
		resultChan <- result
	})

	// 等待窗口触发（处理时间模式）
	//fmt.Println("等待窗口初始化...")
	time.Sleep(1 * time.Second)

	// 手动触发窗口
	//fmt.Println("手动触发窗口")
	strm.Window.Trigger()

	// 等待结果
	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()

	var actual interface{}
	select {
	case actual = <-resultChan:
		//fmt.Println("成功接收到结果")
		cancel()
	case <-ctx.Done():
		t.Fatal("测试超时，未收到结果")
	}

	// 验证结果
	resultSlice, ok := actual.([]map[string]interface{})
	require.True(t, ok, "结果应该是[]map[string]interface{}类型")

	// 验证结果数量
	assert.Len(t, resultSlice, 2, "应该有2个设备的聚合结果")

	// 检查设备及其聚合结果
	for _, result := range resultSlice {
		device, _ := result["device"].(string)
		total := result["total"]
		average := result["average"]
		count := result["cnt"]

		//fmt.Printf("设备 %s: total=%v, average=%v, count=%v\n", device, total, average, count)

		if device == "sensor1" {
			// sensor1: sum=60, avg=20, count=3
			if total != nil {
				assert.Equal(t, 60.0, total)
			}
			if average != nil {
				assert.Equal(t, 20.0, average)
			}
			if count != nil {
				assert.Equal(t, 3.0, count)
			}
		} else if device == "sensor2" {
			// sensor2: sum=40, avg=20, count=2
			if total != nil {
				assert.Equal(t, 40.0, total)
			}
			if average != nil {
				assert.Equal(t, 20.0, average)
			}
			if count != nil {
				assert.Equal(t, 2.0, count)
			}
		}
	}

	//fmt.Println("基本增量计算测试完成")
}

// TestExprFunctions 测试expr函数的使用
func TestExprFunctions(t *testing.T) {
	streamsql := New()
	defer streamsql.Stop()

	// 测试基本expr函数：字符串处理
	var rsql = "SELECT device, upper(device) as upper_device, lower(device) as lower_device FROM stream"
	err := streamsql.Execute(rsql)
	assert.Nil(t, err)
	strm := streamsql.stream

	// 创建结果接收通道
	resultChan := make(chan interface{}, 10)

	// 添加结果回调
	strm.AddSink(func(result []map[string]interface{}) {
		resultChan <- result
	})

	// 添加测试数据
	testData := []map[string]interface{}{
		{"device": "SensorA"},
		{"device": "SensorB"},
	}

	// 添加数据
	for _, data := range testData {
		strm.Emit(data)
	}

	// 等待结果
	var results []interface{}
	var resultsMutex sync.Mutex
	timeout := time.After(2 * time.Second)
	done := false

	for !done {
		resultsMutex.Lock()
		resultCount := len(results)
		resultsMutex.Unlock()

		if resultCount >= 2 {
			break
		}

		select {
		case result := <-resultChan:
			resultsMutex.Lock()
			results = append(results, result)
			resultsMutex.Unlock()
		case <-timeout:
			done = true
		}
	}

	// 验证结果
	resultsMutex.Lock()
	finalResultCount := len(results)
	resultsCopy := make([]interface{}, len(results))
	copy(resultsCopy, results)
	resultsMutex.Unlock()

	assert.Greater(t, finalResultCount, 0, "应该收到至少一条结果")

	for _, result := range resultsCopy {
		resultSlice, ok := result.([]map[string]interface{})
		require.True(t, ok, "结果应该是[]map[string]interface{}类型")

		for _, item := range resultSlice {
			device, _ := item["device"].(string)
			upperDevice, _ := item["upper_device"].(string)
			lowerDevice, _ := item["lower_device"].(string)

			// 验证upper函数
			assert.Equal(t, strings.ToUpper(device), upperDevice, "upper函数应该正确转换大写")
			// 验证lower函数
			assert.Equal(t, strings.ToLower(device), lowerDevice, "lower函数应该正确转换小写")
		}
	}
}

// TestExprFunctionsInAggregation 测试在聚合中使用expr函数
func TestExprFunctionsInAggregation(t *testing.T) {
	streamsql := New()
	defer streamsql.Stop()

	// 测试在聚合函数中使用expr函数：数学计算
	var rsql = "SELECT device, AVG(abs(temperature - 25)) as avg_deviation, MAX(ceil(temperature)) as max_ceil FROM stream GROUP BY device, TumblingWindow('1s')"
	err := streamsql.Execute(rsql)
	assert.Nil(t, err)
	strm := streamsql.stream

	// 添加测试数据
	// 不使用事件时间，不需要时间戳字段
	testData := []map[string]interface{}{
		{"device": "sensor1", "temperature": 23.5}, // abs(23.5-25) = 1.5, ceil(23.5) = 24
		{"device": "sensor1", "temperature": 26.8}, // abs(26.8-25) = 1.8, ceil(26.8) = 27
		{"device": "sensor2", "temperature": 24.2}, // abs(24.2-25) = 0.8, ceil(24.2) = 25
		{"device": "sensor2", "temperature": 25.9}, // abs(25.9-25) = 0.9, ceil(25.9) = 26
	}

	// 创建结果接收通道
	resultChan := make(chan interface{}, 10)

	// 添加结果回调
	strm.AddSink(func(result []map[string]interface{}) {
		resultChan <- result
	})

	// 添加数据
	for _, data := range testData {
		strm.Emit(data)
	}

	// 等待窗口初始化
	time.Sleep(1 * time.Second)

	// 手动触发窗口
	strm.Window.Trigger()

	// 等待结果
	ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
	defer cancel()

	var actual interface{}
	select {
	case actual = <-resultChan:
		cancel()
	case <-ctx.Done():
		t.Fatal("测试超时，未收到结果")
	}

	// 验证结果
	resultSlice, ok := actual.([]map[string]interface{})
	require.True(t, ok, "结果应该是[]map[string]interface{}类型")
	assert.Len(t, resultSlice, 2, "应该有2个设备的聚合结果")

	// 检查聚合结果
	for _, result := range resultSlice {
		device, _ := result["device"].(string)
		avgDeviation, ok := result["avg_deviation"].(float64)
		assert.True(t, ok, "avg_deviation应该是float64类型")
		maxCeil, ok := result["max_ceil"].(float64)
		assert.True(t, ok, "max_ceil应该是float64类型")

		if device == "sensor1" {
			// sensor1: avg(abs(23.5-25), abs(26.8-25)) = avg(1.5, 1.8) = 1.65
			assert.InEpsilon(t, 1.65, avgDeviation, 0.01, "sensor1的平均偏差应为1.65")
			// sensor1: max(ceil(23.5), ceil(26.8)) = max(24, 27) = 27
			assert.Equal(t, 27.0, maxCeil, "sensor1的最大向上取整应为27")
		} else if device == "sensor2" {
			// sensor2: avg(abs(24.2-25), abs(25.9-25)) = avg(0.8, 0.9) = 0.85
			assert.InEpsilon(t, 0.85, avgDeviation, 0.01, "sensor2的平均偏差应为0.85")
			// sensor2: max(ceil(24.2), ceil(25.9)) = max(25, 26) = 26
			assert.Equal(t, 26.0, maxCeil, "sensor2的最大向上取整应为26")
		}
	}
}

// TestNestedExprFunctions 测试嵌套expr函数调用
func TestNestedExprFunctions(t *testing.T) {
	streamsql := New()
	defer streamsql.Stop()

	// 测试嵌套函数：字符串处理 + 数组操作
	var rsql = "SELECT device, len(split(upper(device), 'SENSOR')) as split_count FROM stream"
	err := streamsql.Execute(rsql)
	assert.Nil(t, err)
	strm := streamsql.stream

	// 创建结果接收通道
	resultChan := make(chan interface{}, 10)

	// 添加结果回调
	strm.AddSink(func(result []map[string]interface{}) {
		resultChan <- result
	})

	// 添加测试数据
	testData := []map[string]interface{}{
		{"device": "sensor1"},      // upper -> "SENSOR1", split by "SENSOR" -> ["", "1"], len -> 2
		{"device": "sensorsensor"}, // upper -> "SENSORSENSOR", split by "SENSOR" -> ["", "", ""], len -> 3
		{"device": "device1"},      // upper -> "DEVICE1", split by "SENSOR" -> ["DEVICE1"], len -> 1
	}

	// 添加数据
	for _, data := range testData {
		strm.Emit(data)
	}

	// 等待结果
	var results []interface{}
	var resultsMutex sync.Mutex
	timeout := time.After(2 * time.Second)
	done := false

	for !done {
		resultsMutex.Lock()
		resultCount := len(results)
		resultsMutex.Unlock()

		if resultCount >= 3 {
			break
		}

		select {
		case result := <-resultChan:
			resultsMutex.Lock()
			results = append(results, result)
			resultsMutex.Unlock()
		case <-timeout:
			done = true
		}
	}

	// 验证结果
	resultsMutex.Lock()
	finalResultCount := len(results)
	resultsCopy := make([]interface{}, len(results))
	copy(resultsCopy, results)
	resultsMutex.Unlock()

	assert.Greater(t, finalResultCount, 0, "应该收到至少一条结果")

	deviceResults := make(map[string]float64)
	for _, result := range resultsCopy {
		resultSlice, ok := result.([]map[string]interface{})
		require.True(t, ok, "结果应该是[]map[string]interface{}类型")

		for _, item := range resultSlice {
			device, _ := item["device"].(string)
			splitCount, ok := item["split_count"].(float64)
			if ok {
				deviceResults[device] = splitCount
			}
		}
	}

	// 验证嵌套函数调用结果
	if count, exists := deviceResults["sensor1"]; exists {
		assert.Equal(t, 2.0, count, "sensor1经过嵌套函数处理后应该得到2")
	}
	if count, exists := deviceResults["sensorsensor"]; exists {
		assert.Equal(t, 3.0, count, "sensorsensor经过嵌套函数处理后应该得到3")
	}
	if count, exists := deviceResults["device1"]; exists {
		assert.Equal(t, 1.0, count, "device1经过嵌套函数处理后应该得到1")
	}
}

// TestExprFunctionsWithStreamSQLFunctions 测试expr函数与StreamSQL函数混合使用
func TestExprFunctionsWithStreamSQLFunctions(t *testing.T) {
	streamsql := New()
	defer streamsql.Stop()

	// 测试混合使用：StreamSQL的concat函数 + expr的upper函数
	var rsql = "SELECT device, concat(upper(device), '_processed') as processed_name FROM stream"
	err := streamsql.Execute(rsql)
	assert.Nil(t, err)
	strm := streamsql.stream

	// 创建结果接收通道
	resultChan := make(chan interface{}, 10)

	// 添加结果回调
	strm.AddSink(func(result []map[string]interface{}) {
		resultChan <- result
	})

	// 添加测试数据
	testData := []map[string]interface{}{
		{"device": "sensor1"},
		{"device": "device2"},
	}

	// 添加数据
	for _, data := range testData {
		strm.Emit(data)
	}

	// 等待结果
	var results []interface{}
	var resultsMutex sync.Mutex
	timeout := time.After(2 * time.Second)
	done := false

	for !done {
		resultsMutex.Lock()
		resultCount := len(results)
		resultsMutex.Unlock()

		if resultCount >= 2 {
			break
		}

		select {
		case result := <-resultChan:
			resultsMutex.Lock()
			results = append(results, result)
			resultsMutex.Unlock()
		case <-timeout:
			done = true
		}
	}

	// 验证结果
	resultsMutex.Lock()
	finalResultCount := len(results)
	resultsCopy := make([]interface{}, len(results))
	copy(resultsCopy, results)
	resultsMutex.Unlock()

	assert.Greater(t, finalResultCount, 0, "应该收到至少一条结果")

	for _, result := range resultsCopy {
		resultSlice, ok := result.([]map[string]interface{})
		require.True(t, ok, "结果应该是[]map[string]interface{}类型")

		for _, item := range resultSlice {
			device, _ := item["device"].(string)
			processedName, _ := item["processed_name"].(string)

			// 验证混合函数调用结果
			expected := strings.ToUpper(device) + "_processed"
			assert.Equal(t, expected, processedName, "混合函数调用应该正确处理")
		}
	}
}

// TestSelectAllFeature 专门测试SELECT *功能
func TestSelectAllFeature(t *testing.T) {
	// 测试场景1：基本SELECT *查询
	t.Run("基本SELECT *查询", func(t *testing.T) {
		streamsql := New()
		defer streamsql.Stop()

		var rsql = "SELECT * FROM stream"
		err := streamsql.Execute(rsql)
		assert.Nil(t, err)
		strm := streamsql.stream

		// 创建结果接收通道
		resultChan := make(chan interface{}, 10)

		// 添加结果接收器
		strm.AddSink(func(result []map[string]interface{}) {
			resultChan <- result
		})

		// 添加测试数据
		testData := map[string]interface{}{
			"device":      "sensor001",
			"temperature": 25.5,
			"humidity":    60,
			"location":    "room1",
			"status":      "active",
		}

		// 发送数据
		strm.Emit(testData)

		// 等待结果
		ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
		defer cancel()

		select {
		case result := <-resultChan:
			// 验证结果
			resultSlice, ok := result.([]map[string]interface{})
			require.True(t, ok, "结果应该是[]map[string]interface{}类型")
			require.Len(t, resultSlice, 1, "应该只有一条结果")

			item := resultSlice[0]
			// 验证所有原始字段都存在
			assert.Equal(t, "sensor001", item["device"], "device字段应该正确")
			assert.Equal(t, 25.5, item["temperature"], "temperature字段应该正确")
			assert.Equal(t, 60, item["humidity"], "humidity字段应该正确")
			assert.Equal(t, "room1", item["location"], "location字段应该正确")
			assert.Equal(t, "active", item["status"], "status字段应该正确")

			// 验证字段数量
			assert.Len(t, item, 5, "应该包含所有5个字段")

			cancel()
		case <-ctx.Done():
			t.Fatal("测试超时，未收到结果")
		}
	})

	// 测试场景2：SELECT * + WHERE条件
	t.Run("SELECT * + WHERE条件", func(t *testing.T) {
		streamsql := New()
		defer streamsql.Stop()

		var rsql = "SELECT * FROM stream WHERE temperature > 20"
		err := streamsql.Execute(rsql)
		assert.Nil(t, err)
		strm := streamsql.stream

		// 创建结果接收通道
		resultChan := make(chan interface{}, 10)

		// 添加结果接收器
		strm.AddSink(func(result []map[string]interface{}) {
			resultChan <- result
		})

		// 添加测试数据
		testData := []map[string]interface{}{
			{"device": "sensor1", "temperature": 25.0, "humidity": 60}, // 应该被包含
			{"device": "sensor2", "temperature": 15.0, "humidity": 70}, // 应该被过滤掉
			{"device": "sensor3", "temperature": 30.0, "humidity": 50}, // 应该被包含
		}

		var results []interface{}
		var resultsMutex sync.Mutex

		// 发送数据
		for _, data := range testData {
			strm.Emit(data)

			// 立即检查结果
			ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
			select {
			case result := <-resultChan:
				resultsMutex.Lock()
				results = append(results, result)
				resultsMutex.Unlock()
				cancel()
			case <-ctx.Done():
				cancel()
				// 对于不满足条件的数据，超时是正常的
			}
		}

		// 验证结果
		resultsMutex.Lock()
		finalResultCount := len(results)
		resultsCopy := make([]interface{}, len(results))
		copy(resultsCopy, results)
		resultsMutex.Unlock()

		assert.Equal(t, 2, finalResultCount, "应该有2条记录满足条件")

		// 验证结果内容
		deviceFound := make(map[string]bool)
		for _, result := range resultsCopy {
			resultSlice, ok := result.([]map[string]interface{})
			require.True(t, ok, "结果应该是[]map[string]interface{}类型")
			require.Len(t, resultSlice, 1, "每个结果应该只有一条记录")

			item := resultSlice[0]
			device, _ := item["device"].(string)
			temp, _ := item["temperature"].(float64)

			// 验证温度条件
			assert.Greater(t, temp, 20.0, "温度应该大于20")

			// 记录找到的设备
			deviceFound[device] = true

			// 验证所有字段都存在
			assert.Contains(t, item, "device", "应该包含device字段")
			assert.Contains(t, item, "temperature", "应该包含temperature字段")
			assert.Contains(t, item, "humidity", "应该包含humidity字段")
		}

		// 验证正确的设备被包含
		assert.True(t, deviceFound["sensor1"], "sensor1应该被包含")
		assert.True(t, deviceFound["sensor3"], "sensor3应该被包含")
		assert.False(t, deviceFound["sensor2"], "sensor2不应该被包含")
	})

	// 测试场景3：SELECT * + LIMIT
	t.Run("SELECT * + LIMIT", func(t *testing.T) {
		streamsql := New()
		defer streamsql.Stop()

		var rsql = "SELECT * FROM stream LIMIT 2"
		err := streamsql.Execute(rsql)
		assert.Nil(t, err)
		strm := streamsql.stream

		// 创建结果接收通道
		resultChan := make(chan interface{}, 10)

		// 添加结果接收器
		strm.AddSink(func(result []map[string]interface{}) {
			resultChan <- result
		})

		// 添加测试数据
		testData := []map[string]interface{}{
			{"device": "sensor1", "temperature": 25.0},
			{"device": "sensor2", "temperature": 26.0},
			{"device": "sensor3", "temperature": 27.0},
			{"device": "sensor4", "temperature": 28.0},
		}

		var results []interface{}
		var resultsMutex sync.Mutex

		// 发送数据
		for _, data := range testData {
			strm.Emit(data)

			// 立即检查结果
			ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
			select {
			case result := <-resultChan:
				resultsMutex.Lock()
				results = append(results, result)
				resultsMutex.Unlock()
				cancel()
			case <-ctx.Done():
				cancel()
			}
		}

		// 验证结果
		resultsMutex.Lock()
		finalResultCount := len(results)
		resultsCopy := make([]interface{}, len(results))
		copy(resultsCopy, results)
		resultsMutex.Unlock()

		assert.GreaterOrEqual(t, finalResultCount, 2, "应该至少有2条结果")

		// 验证结果内容
		for _, result := range resultsCopy {
			resultSlice, ok := result.([]map[string]interface{})
			require.True(t, ok, "结果应该是[]map[string]interface{}类型")

			// 验证LIMIT限制：每个batch最多2条记录
			assert.LessOrEqual(t, len(resultSlice), 2, "每个batch最多2条记录")
			assert.Greater(t, len(resultSlice), 0, "应该有结果")

			// 验证字段
			for _, item := range resultSlice {
				assert.Contains(t, item, "device", "结果应包含device字段")
				assert.Contains(t, item, "temperature", "结果应包含temperature字段")
			}
		}
	})

	// 测试场景4：SELECT * with嵌套字段
	t.Run("SELECT * with嵌套字段", func(t *testing.T) {
		streamsql := New()
		defer streamsql.Stop()

		var rsql = "SELECT * FROM stream"
		err := streamsql.Execute(rsql)
		assert.Nil(t, err)
		strm := streamsql.stream

		// 创建结果接收通道
		resultChan := make(chan interface{}, 10)

		// 添加结果接收器
		strm.AddSink(func(result []map[string]interface{}) {
			resultChan <- result
		})

		// 添加带嵌套字段的测试数据
		testData := map[string]interface{}{
			"device": "sensor001",
			"metrics": map[string]interface{}{
				"temperature": 25.5,
				"humidity":    60,
			},
			"location": map[string]interface{}{
				"building": "A",
				"room":     "101",
			},
		}

		// 发送数据
		strm.Emit(testData)

		// 等待结果
		ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
		defer cancel()

		select {
		case result := <-resultChan:
			// 验证结果
			resultSlice, ok := result.([]map[string]interface{})
			require.True(t, ok, "结果应该是[]map[string]interface{}类型")
			require.Len(t, resultSlice, 1, "应该只有一条结果")

			item := resultSlice[0]
			// 验证顶级字段
			assert.Equal(t, "sensor001", item["device"], "device字段应该正确")

			// 验证嵌套字段结构被保留
			metrics, ok := item["metrics"].(map[string]interface{})
			assert.True(t, ok, "metrics应该是map类型")
			assert.Equal(t, 25.5, metrics["temperature"], "嵌套temperature字段应该正确")
			assert.Equal(t, 60, metrics["humidity"], "嵌套humidity字段应该正确")

			location, ok := item["location"].(map[string]interface{})
			assert.True(t, ok, "location应该是map类型")
			assert.Equal(t, "A", location["building"], "嵌套building字段应该正确")
			assert.Equal(t, "101", location["room"], "嵌套room字段应该正确")

			cancel()
		case <-ctx.Done():
			t.Fatal("测试超时，未收到结果")
		}
	})
}

// TestCaseNullValueHandlingInAggregation 测试CASE表达式在聚合函数中正确处理NULL值
func TestCaseNullValueHandlingInAggregation(t *testing.T) {
	sql := `SELECT deviceType,
	              SUM(CASE WHEN temperature > 30 THEN temperature ELSE NULL END) as high_temp_sum,
	              COUNT(CASE WHEN temperature > 30 THEN 1 ELSE NULL END) as high_temp_count,
	              AVG(CASE WHEN temperature > 30 THEN temperature ELSE NULL END) as high_temp_avg
	         FROM stream 
	         GROUP BY deviceType, TumblingWindow('2s')`

	// 创建StreamSQL实例
	ssql := New()
	defer ssql.Stop()

	// 执行SQL
	err := ssql.Execute(sql)
	require.NoError(t, err)

	// 收集结果
	var results []map[string]interface{}
	resultChan := make(chan interface{}, 10)

	ssql.AddSink(func(result []map[string]interface{}) {
		// 非阻塞发送，避免阻塞 sink worker
		select {
		case resultChan <- result:
		default:
			// Channel 已满，忽略（非阻塞发送）
		}
	})

	// 添加测试数据
	testData := []map[string]interface{}{
		{"deviceType": "sensor", "temperature": 35.0},  // 满足条件
		{"deviceType": "sensor", "temperature": 25.0},  // 不满足条件，返回NULL
		{"deviceType": "sensor", "temperature": 32.0},  // 满足条件
		{"deviceType": "monitor", "temperature": 28.0}, // 不满足条件，返回NULL
		{"deviceType": "monitor", "temperature": 33.0}, // 满足条件
	}

	for _, data := range testData {
		ssql.Emit(data)
	}

	// 等待窗口触发
	time.Sleep(3 * time.Second)

	// 收集结果（添加超时机制）
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

collecting:
	for {
		select {
		case result := <-resultChan:
			if resultSlice, ok := result.([]map[string]interface{}); ok {
				results = append(results, resultSlice...)
			}
		case <-time.After(500 * time.Millisecond):
			break collecting
		case <-ctx.Done():
			break collecting
		}
	}

	// 等待一小段时间，确保所有 sink worker 完成当前任务
	// 这样可以确保所有结果都被发送到 channel
	time.Sleep(100 * time.Millisecond)

	// 取消 context，通知所有相关 goroutine 退出
	// 不关闭 resultChan，让主程序自动退出时清理
	cancel()

	// 验证结果
	assert.Len(t, results, 2, "应该有两个设备类型的结果")

	// 验证各个deviceType的结果
	expectedResults := map[string]map[string]interface{}{
		"sensor": {
			"high_temp_sum":   67.0, // 35 + 32
			"high_temp_count": 2.0,  // COUNT应该忽略NULL
			"high_temp_avg":   33.5, // (35 + 32) / 2
		},
		"monitor": {
			"high_temp_sum":   33.0, // 只有33
			"high_temp_count": 1.0,  // COUNT应该忽略NULL
			"high_temp_avg":   33.0, // 只有33
		},
	}

	for _, result := range results {
		deviceType := result["deviceType"].(string)
		expected := expectedResults[deviceType]

		assert.NotNil(t, expected, "应该有设备类型 %s 的期望结果", deviceType)

		// 验证SUM聚合（忽略NULL值）
		assert.Equal(t, expected["high_temp_sum"], result["high_temp_sum"],
			"设备类型 %s 的SUM聚合结果应该正确", deviceType)

		// 验证COUNT聚合（忽略NULL值）
		assert.Equal(t, expected["high_temp_count"], result["high_temp_count"],
			"设备类型 %s 的COUNT聚合结果应该正确", deviceType)

		// 验证AVG聚合（忽略NULL值）
		assert.Equal(t, expected["high_temp_avg"], result["high_temp_avg"],
			"设备类型 %s 的AVG聚合结果应该正确", deviceType)
	}
}