test: 合并 coverage/perf/strategy/table_print 到 streamsql_test.go，根目录测试文件收敛到 3 个（options/benchmark/streamsql_test）

streamsql_perf_test.go

@@ -1,69 +0,0 @@
-package streamsql
-
-import (
-	"testing"
-)
-
-// Integration benchmarks exercising the full main path with realistic RSQL.
-// EmitSync processes each row synchronously end-to-end (the same path users
-// call), so ns/op is the true per-row latency through ProcessData -> field
-// evaluation -> result building. Aggregation queries are exercised separately
-// via the Emit-based benchmarks.
-
-func benchEmitSync(b *testing.B, sql string, row map[string]interface{}) {
-	b.Helper()
-	ssql := New()
-	defer ssql.Stop()
-	if err := ssql.Execute(sql); err != nil {
-		b.Fatalf("Execute: %v", err)
-	}
-
-	// Warm up compile/preprocess caches (do not measure).
-	if _, err := ssql.EmitSync(row); err != nil {
-		b.Fatalf("warmup EmitSync: %v", err)
-	}
-
-	b.ReportAllocs()
-	b.ResetTimer()
-	for i := 0; i < b.N; i++ {
-		if _, err := ssql.EmitSync(row); err != nil {
-			b.Fatalf("EmitSync: %v", err)
-		}
-	}
-	b.StopTimer()
-}
-
-func BenchmarkMainPath_FilterProject(b *testing.B) {
-	benchEmitSync(b,
-		"SELECT deviceId, temperature FROM stream WHERE temperature > 20",
-		map[string]interface{}{"deviceId": "d1", "temperature": 25.5, "humidity": 60.0},
-	)
-}
-
-func BenchmarkMainPath_MultiFieldFilter(b *testing.B) {
-	benchEmitSync(b,
-		"SELECT deviceId, temperature, humidity FROM stream WHERE temperature > 20 AND humidity < 80",
-		map[string]interface{}{"deviceId": "d1", "temperature": 25.5, "humidity": 60.0},
-	)
-}
-
-func BenchmarkMainPath_ComputedFields(b *testing.B) {
-	benchEmitSync(b,
-		"SELECT deviceId, temperature * 2 + humidity AS score, abs(temperature - 100) AS dev FROM stream WHERE temperature > 20",
-		map[string]interface{}{"deviceId": "d1", "temperature": 25.5, "humidity": 60.0},
-	)
-}
-
-func BenchmarkMainPath_StringConcat(b *testing.B) {
-	benchEmitSync(b,
-		"SELECT deviceId + '-' + location AS id FROM stream",
-		map[string]interface{}{"deviceId": "d1", "location": "roomA"},
-	)
-}
-
-func BenchmarkMainPath_NoFilter(b *testing.B) {
-	benchEmitSync(b,
-		"SELECT deviceId, temperature, humidity FROM stream",
-		map[string]interface{}{"deviceId": "d1", "temperature": 25.5, "humidity": 60.0},
-	)
-}

streamsql_strategy_test.go

@@ -1,150 +0,0 @@
-package streamsql
-
-import (
-	"testing"
-	"time"
-
-	"github.com/rulego/streamsql/types"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-)
-
-// TestSQLIntegration_StrategyBlock 测试 SQL 集成下的阻塞策略
-func TestSQLIntegration_StrategyBlock(t *testing.T) {
-	// 配置：输出缓冲为 1，阻塞策略，超时 100ms
-	ssql := New(WithCustomPerformance(types.PerformanceConfig{
-		BufferConfig: types.BufferConfig{
-			DataChannelSize:   100,
-			ResultChannelSize: 100,
-			WindowOutputSize:  1,
-		},
-		OverflowConfig: types.OverflowConfig{
-			Strategy:      types.OverflowStrategyBlock,
-			BlockTimeout:  100 * time.Millisecond,
-			AllowDataLoss: true,
-		},
-		WorkerConfig: types.WorkerConfig{
-			SinkPoolSize:    0, // 无缓冲任务队列
-			SinkWorkerCount: 1, // 1个 worker
-		},
-	}))
-	defer ssql.Stop()
-
-	// SQL: 每条数据触发一次窗口
-	rsql := "SELECT deviceId FROM stream GROUP BY deviceId, CountingWindow(1)"
-	err := ssql.Execute(rsql)
-	require.NoError(t, err)
-
-	// 添加同步 Sink 阻塞 Stream 处理，从而反压 Window
-	// 注意：必须在 Execute 之后添加，因为 Execute 才会创建 stream
-	ssql.AddSyncSink(func(results []map[string]interface{}) {
-		time.Sleep(500 * time.Millisecond)
-	})
-
-	// 发送 5 条数据
-	// d1: Worker 处理中 (阻塞 500ms)
-	// d2: Stream 尝试写入 WorkerPool -> 阻塞 (无缓冲)
-	// d3: Window OutputChan (size 1) -> 填满
-	// d4: Window OutputChan 满 -> 尝试写入 -> 阻塞 (Window Add) -> 放入 TriggerChan (size=1)
-	// d5: Window Add -> TriggerChan 满 -> 阻塞? No, Emit 是异步的?
-	// Emit 往 dataChan 写. DataProcessor 读 dataChan -> Window.Add.
-	// Window.Add 往 triggerChan 写.
-	//
-	// 修正分析:
-	// Window.Add 是非阻塞的 (如果 triggerChan 不满).
-	// CountingWindow triggerChan size = bufferSize = 1.
-	// Worker 协程: 从 triggerChan 读 -> 处理 -> sendResult (到 OutputChan).
-	//
-	// d1: Worker读triggerChan -> OutputChan -> Stream -> WorkerPool -> Worker(busy).
-	// d2: Worker读triggerChan -> OutputChan -> Stream -> Blocked on WorkerPool.
-	//     此时 Stream 持有 d2. OutputChan 空.
-	//     Worker 协程 阻塞在 sendResult(d2)? No, Stream 取走了 d2, Stream 阻塞在 dispatch.
-	//     所以 OutputChan 是空的!
-	//     Wait, Stream loop:
-	//     result := <-OutputChan. (Stream has d2).
-	//     handleResult(d2) -> Blocked.
-	//     So OutputChan is empty.
-	// d3: Worker读triggerChan -> OutputChan (d3). Success.
-	//     OutputChan has d3.
-	// d4: Worker读triggerChan -> OutputChan (d4). Blocked (OutputChan full).
-	//     Worker 协程 阻塞在 sendResult(d4).
-	// d5: Add -> triggerChan (d5). Success (triggerChan size 1).
-	// d6: Add -> triggerChan (d6). Blocked (triggerChan full).
-	//     Add blocks. DataProcessor blocks. Emit succeeds (dataChan).
-	//
-	// 所以 Window Worker 只有在 sendResult 阻塞时才触发 Drop logic.
-	// sendResult 只有在 OutputChan 满且超时时才 Drop.
-	//
-	// d4 阻塞在 sendResult.
-	// 100ms 后超时 -> Drop d4.
-	// Worker 继续.
-	//
-	// 所以 d4 应该是被 Drop 的那个.
-	// Sent: d1, d2, d3. (d5 在 triggerChan, d6 在 dataChan).
-	// Wait, d5 is in triggerChan, not processed yet.
-	// So Sent = 3. Dropped = 1 (d4).
-
-	for _, id := range []string{"d1", "d2", "d3", "d4", "d5"} {
-		ssql.Emit(map[string]interface{}{"deviceId": id})
-		time.Sleep(10 * time.Millisecond)
-	}
-
-	// 等待足够长的时间让 Stream 醒来并处理完，以及 Window 丢弃逻辑执行
-	time.Sleep(1000 * time.Millisecond)
-
-	// 获取统计信息
-	// d1: Stream 处理完
-	// d2: Stream 处理完 (Worker 醒来后处理 d2)
-	// d3: Dropped (Worker 阻塞 -> 超时)
-	// d4: Dropped (Worker 阻塞 -> 超时)
-	// d5: Dropped (Worker 阻塞 -> 超时)
-	// Total Sent: 2 (d1, d2).
-	// Dropped: 3 (d3, d4, d5).
-	stats := ssql.stream.GetStats()
-	assert.Equal(t, int64(3), stats["droppedCount"], "Should have 3 dropped window result due to overflow")
-	assert.Equal(t, int64(2), stats["sentCount"], "Should have 2 sent window result")
-}
-
-// TestSQLIntegration_StrategyDrop 测试 SQL 集成下的丢弃策略
-func TestSQLIntegration_StrategyDrop(t *testing.T) {
-	// 配置：输出缓冲为 1，丢弃策略
-	ssql := New(WithCustomPerformance(types.PerformanceConfig{
-		BufferConfig: types.BufferConfig{
-			DataChannelSize:   100,
-			ResultChannelSize: 100,
-			WindowOutputSize:  1,
-		},
-		OverflowConfig: types.OverflowConfig{
-			Strategy: types.OverflowStrategyDrop,
-		},
-	}))
-	defer ssql.Stop()
-
-	// SQL: 每条数据触发一次窗口
-	rsql := "SELECT deviceId FROM stream GROUP BY deviceId, CountingWindow(1)"
-	err := ssql.Execute(rsql)
-	require.NoError(t, err)
-
-	// 连续发送 3 条数据
-	ssql.Emit(map[string]interface{}{"deviceId": "d1"})
-	ssql.Emit(map[string]interface{}{"deviceId": "d2"})
-	ssql.Emit(map[string]interface{}{"deviceId": "d3"})
-
-	// 等待处理完成
-	time.Sleep(200 * time.Millisecond)
-
-	// 对于 StrategyDrop，它会挤掉旧数据，所以 sentCount 应该持续增加
-	stats := ssql.stream.GetStats()
-	// d1, d2, d3 都会成功发送（虽然 d1, d2 可能被挤掉，但 sendResult 逻辑中挤掉旧的后写入新的算发送成功）
-	assert.Equal(t, int64(3), stats["sentCount"])
-
-	// 验证最终留在缓冲区的是最后一条数据 (d3)
-	// 注意：AddSink 会启动 worker 从 OutputChan 读。
-	// 为了验证，我们直接从 Window 的 OutputChan 读
-	select {
-	case result := <-ssql.stream.Window.OutputChan():
-		assert.Equal(t, "d3", result[0].Data.(map[string]interface{})["deviceId"])
-	case <-time.After(100 * time.Millisecond):
-		// 如果已经被 AddSink 的 worker 读走了也正常，但由于我们没加 Sink，所以应该在里面
-	}
-}

streamsql_table_print_test.go

@@ -1,45 +0,0 @@
-package streamsql
-
-import (
-	"testing"
-	"time"
-
-	"github.com/stretchr/testify/assert"
-)
-
-// TestPrintTable 测试PrintTable方法的基本功能
-func TestPrintTable(t *testing.T) {
-	// 创建StreamSQL实例并测试PrintTable
-	ssql := New()
-	err := ssql.Execute("SELECT device, AVG(temperature) as avg_temp FROM stream GROUP BY device, TumblingWindow('2s')")
-	assert.NoError(t, err)
-
-	// 使用PrintTable方法（不验证输出内容，只确保不会panic）
-	assert.NotPanics(t, func() {
-		ssql.PrintTable()
-	}, "PrintTable方法不应该panic")
-
-	// 发送测试数据
-	testData := []map[string]interface{}{
-		{"device": "sensor1", "temperature": 25.0},
-		{"device": "sensor2", "temperature": 30.0},
-	}
-
-	for _, data := range testData {
-		ssql.Emit(data)
-	}
-
-	// 等待窗口触发
-	time.Sleep(3 * time.Second)
-}
-
-// TestPrintTableFormat 测试printTableFormat方法处理不同数据类型
-func TestPrintTableFormat(t *testing.T) {
-	ssql := New()
-
-	// 测试不同类型的数据，确保不会panic
-	assert.NotPanics(t, func() {
-		// 测试空切片
-		ssql.printTableFormat([]map[string]interface{}{})
-	}, "空切片不应该panic")
-}