feat: support state window (#3757)

Signed-off-by: Song Gao <disxiaofei@163.com>

Author: Yisaer

docs/en_US/sqls/windows.md

@@ -72,6 +72,17 @@ A session window begins when the first event occurs. If another event occurs wit
 
 If events keep occurring within the specified timeout, the session window will keep extending until maximum duration is reached. The maximum duration checking intervals are set to be the same size as the specified max duration. For example, if the max duration is 10, then the checks on if the window exceed maximum duration will happen at t = 0, 10, 20, 30, etc.
 
+## Conditional state window
+
+The conditional state window does not focus on time, but only on the impact of each piece of data on the window state. It has two main parameters, the start window trigger condition and the send window trigger condition.
+
+```sql
+SELECT * from demo group by statewindow(a > 1, a > 5)
+```
+
+The conditional state window is initially in the untriggered state, at which point all data entering the window will be directly discarded. When the data entering the window meets the start window trigger condition, it changes from the untriggered state to the triggered state, at which point all data entering the window will be stored.
+When the conditional state window is in the triggered state, when the data entering the window meets the send window trigger condition, it changes from the triggered state to the untriggered state, and all data previously stored in the window will be sent as a window.
+
 ## Count window
 
 Please notice that the count window does not concern time, it only concern about events count.

docs/zh_CN/sqls/windows.md

@@ -72,6 +72,17 @@ SELECT count(*) FROM demo GROUP BY ID, SESSIONWINDOW(mi, 2, 1);
 
 如果事件在指定的超时时间内持续发生，则会话窗口将继续扩展直到达到最大持续时间。 最大持续时间检查间隔设置为与指定的最大持续时间相同的大小。 例如，如果最大持续时间为10，则检查窗口是否超过最大持续时间将在 t = 0、10、20、30等处进行。
 
+## 条件状态窗口
+
+条件状态窗口不关注时间，只关注每条数据对窗口状态的影响。他有两个主要参数, 开始窗口触发条件与发送窗口触发条件。
+
+```sql
+SELECT * from demo group by statewindow(a > 1, a > 5)
+```
+
+条件状态窗口最初是未触发状态，此时所有进入窗口的数据都会被直接丢弃。当进入窗口的数据满足开始窗口触发条件时，则从未触发状态转变为触发状态，此时所有进入窗口的数据都会被存储。
+条件状态窗口处于触发状态时，当进入窗口的数据满足发送窗口触发条件时，则从触发状态转变为未触发状态，且此前所有被存储在窗口的数据都会被作为一个窗口发送。
+
 ## 计数窗口
 
 请注意计数窗口不关注时间，只关注事件发生的次数。

internal/topo/node/window_op.go

@@ -36,6 +36,8 @@ import (
 )
 
 type WindowConfig struct {
+	BeginCondition   ast.Expr
+	EmitCondition    ast.Expr
 	TriggerCondition ast.Expr
 	StateFuncs       []*ast.Call
 	Type             ast.WindowType

internal/topo/node/window_v2_op.go

@@ -27,6 +27,8 @@ import (
 	"github.com/lf-edge/ekuiper/v2/pkg/timex"
 )
 
+var InfTime = time.Unix(1<<63-62135596801, 999999999)
+
 type WindowV2Operator struct {
 	*defaultSinkNode
 	windowConfig WindowConfig
@@ -46,6 +48,8 @@ func NewWindowV2Op(name string, w WindowConfig, options *def.RuleOption) (*Windo
 		} else {
 			o.wExec = NewSlidingWindowOp(o)
 		}
+	case ast.STATE_WINDOW:
+		o.wExec = NewStateWindowOp(o)
 	default:
 		return nil, fmt.Errorf("unsupported window type:%v", w.Type.String())
 	}
@@ -87,6 +91,63 @@ type WindowV2Exec interface {
 	exec(ctx api.StreamContext, errCh chan<- error)
 }
 
+type StateWindowOp struct {
+	*WindowV2Operator
+	BeginCondition ast.Expr
+	EmitCondition  ast.Expr
+	onBegin        bool
+	stateFuncs     []*ast.Call
+}
+
+func NewStateWindowOp(o *WindowV2Operator) *StateWindowOp {
+	return &StateWindowOp{
+		WindowV2Operator: o,
+		BeginCondition:   o.windowConfig.BeginCondition,
+		EmitCondition:    o.windowConfig.EmitCondition,
+		stateFuncs:       o.windowConfig.StateFuncs,
+	}
+}
+
+func (s *StateWindowOp) exec(ctx api.StreamContext, errCh chan<- error) {
+	fv, _ := xsql.NewFunctionValuersForOp(ctx)
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case input := <-s.input:
+			data, processed := s.commonIngest(ctx, input)
+			if processed {
+				continue
+			}
+			s.onProcessStart(ctx, input)
+			switch row := data.(type) {
+			case *xsql.Tuple:
+				var canBegin bool
+				var canEmit bool
+				if !s.onBegin {
+					canBegin = isMatchCondition(ctx, s.BeginCondition, fv, row, s.stateFuncs)
+					if canBegin {
+						s.onBegin = true
+					}
+				}
+				if s.onBegin {
+					s.scanner.addTuple(row)
+					canEmit = isMatchCondition(ctx, s.EmitCondition, fv, row, s.stateFuncs)
+				}
+				if s.onBegin && canEmit {
+					s.emitWindow(ctx, time.Time{}, InfTime)
+					s.scanner.gc(InfTime)
+					s.onBegin = false
+				}
+				if canBegin && !s.onBegin {
+					s.onBegin = true
+				}
+			}
+			s.onProcessEnd(ctx)
+		}
+	}
+}
+
 type SlidingWindowOp struct {
 	*WindowV2Operator
 	Delay            time.Duration
@@ -154,13 +215,13 @@ func (s *SlidingWindowOp) exec(ctx api.StreamContext, errCh chan<- error) {
 	}
 }
 
-func isMatchCondition(ctx api.StreamContext, triggerCondition ast.Expr, fv *xsql.FunctionValuer, d *xsql.Tuple, stateFuncs []*ast.Call) bool {
-	if triggerCondition == nil {
+func isMatchCondition(ctx api.StreamContext, condition ast.Expr, fv *xsql.FunctionValuer, d *xsql.Tuple, stateFuncs []*ast.Call) bool {
+	if condition == nil {
 		return true
 	}
 	log := ctx.GetLogger()
 	ve := &xsql.ValuerEval{Valuer: xsql.MultiValuer(d, fv)}
-	result := ve.Eval(triggerCondition)
+	result := ve.Eval(condition)
 	// not match trigger condition
 	if result == nil {
 		return false

internal/topo/node/window_v2_op_test.go

@@ -37,6 +37,59 @@ func init() {
 	testx.InitEnv("node_test")
 }
 
+func TestStateWindow(t *testing.T) {
+	conf.IsTesting = true
+	now := time.Now()
+	o := &def.RuleOption{
+		BufferLength: 10,
+	}
+	kv, err := store.GetKV("stream")
+	require.NoError(t, err)
+	require.NoError(t, prepareStream())
+	sql := "select count(*) from stream group by statewindow(a>1,a >5)"
+	stmt, err := xsql.NewParser(strings.NewReader(sql)).Parse()
+	require.NoError(t, err)
+	p, err := planner.CreateLogicalPlan(stmt, o, kv)
+	require.NoError(t, err)
+	require.NotNil(t, p)
+	windowPlan := extractWindowPlan(p)
+	require.NotNil(t, windowPlan)
+	op, err := node.NewWindowV2Op("window", node.WindowConfig{
+		Type:           windowPlan.WindowType(),
+		BeginCondition: windowPlan.GetBeginCondition(),
+		EmitCondition:  windowPlan.GetEmitCondition(),
+	}, o)
+	require.NoError(t, err)
+	require.NotNil(t, op)
+	input, _ := op.GetInput()
+	output := make(chan any, 10)
+	op.AddOutput(output, "output")
+	errCh := make(chan error, 10)
+	ctx, cancel := mockContext.NewMockContext("1", "2").WithCancel()
+	op.Exec(ctx, errCh)
+	waitExecute()
+	input <- &xsql.Tuple{Message: map[string]any{"a": int64(1)}, Timestamp: now}
+	input <- &xsql.Tuple{Message: map[string]any{"a": int64(2)}, Timestamp: now.Add(500 * time.Millisecond)}
+	input <- &xsql.Tuple{Message: map[string]any{"a": int64(6)}, Timestamp: now.Add(500 * time.Millisecond)}
+	waitExecute()
+	got := <-output
+	wt, ok := got.(*xsql.WindowTuples)
+	require.True(t, ok)
+	require.NotNil(t, wt)
+	d := wt.ToMaps()
+	require.Equal(t, []map[string]any{
+		{
+			"a": int64(2),
+		},
+		{
+			"a": int64(6),
+		},
+	}, d)
+	cancel()
+	waitExecute()
+	op.Close()
+}
+
 func TestWindowV2SlidingWindowDelay(t *testing.T) {
 	conf.IsTesting = true
 	now := time.Now()

internal/topo/planner/planner.go

@@ -409,6 +409,8 @@ func buildOps(lp LogicalPlan, tp *topo.Topo, options *def.RuleOption, sources ma
 			RawInterval:      rawInterval,
 			TimeUnit:         t.timeUnit,
 			TriggerCondition: t.triggerCondition,
+			BeginCondition:   t.beginCondition,
+			EmitCondition:    t.emitCondition,
 			StateFuncs:       t.stateFuncs,
 		}
 		if options.PlanOptimizeStrategy.GetWindowVersion() == "v2" {
@@ -614,10 +616,14 @@ func createLogicalPlanFull(stmt *ast.SelectStatement, opt *def.RuleOption, store
 				p = incWp
 			} else {
 				wp := WindowPlan{
-					wtype:       w.WindowType,
-					length:      int(w.Length.Val),
-					isEventTime: opt.IsEventTime,
+					wtype:          w.WindowType,
+					isEventTime:    opt.IsEventTime,
+					beginCondition: w.BeginCondition,
+					emitCondition:  w.EmitCondition,
 				}.Init()
+				if w.Length != nil {
+					wp.length = int(w.Length.Val)
+				}
 				if w.Delay != nil {
 					wp.delay = w.Delay.Val
 				}

internal/topo/planner/windowPlan.go

@@ -24,6 +24,8 @@ import (
 
 type WindowPlan struct {
 	baseLogicalPlan
+	beginCondition   ast.Expr
+	emitCondition    ast.Expr
 	triggerCondition ast.Expr
 	condition        ast.Expr
 	wtype            ast.WindowType
@@ -51,6 +53,14 @@ func (p *WindowPlan) GetTriggerCondition() ast.Expr {
 	return p.triggerCondition
 }
 
+func (p *WindowPlan) GetBeginCondition() ast.Expr {
+	return p.beginCondition
+}
+
+func (p *WindowPlan) GetEmitCondition() ast.Expr {
+	return p.emitCondition
+}
+
 func (p *WindowPlan) BuildExplainInfo() {
 	t := p.wtype.String()
 	info := "{ length:" + strconv.Itoa(p.length) + ", "

internal/xsql/parser.go

@@ -873,6 +873,7 @@ var WindowFuncs = map[string]struct{}{
 	"sessionwindow":  {},
 	"slidingwindow":  {},
 	"countwindow":    {},
+	"statewindow":    {},
 	"dedup_trigger":  {},
 }
 
@@ -1045,6 +1046,11 @@ loop:
 
 func validateWindows(fname string, args []ast.Expr) (ast.WindowType, error) {
 	switch fname {
+	case "statewindow":
+		if len(args) != 2 {
+			return ast.STATE_WINDOW, fmt.Errorf("The arguments for %s should be %d.\n", fname, 2)
+		}
+		return ast.STATE_WINDOW, nil
 	case "tumblingwindow":
 		if err := validateWindow(fname, 2, args); err != nil {
 			return ast.TUMBLING_WINDOW, err
@@ -1112,6 +1118,11 @@ func validateWindow(funcName string, expectLen int, args []ast.Expr) error {
 
 func (p *Parser) ConvertToWindows(wtype ast.WindowType, args []ast.Expr) (*ast.Window, error) {
 	win := &ast.Window{WindowType: wtype}
+	if wtype == ast.STATE_WINDOW {
+		win.BeginCondition = args[0]
+		win.EmitCondition = args[1]
+		return win, nil
+	}
 	if wtype == ast.COUNT_WINDOW {
 		win.Length = &ast.IntegerLiteral{Val: args[0].(*ast.IntegerLiteral).Val}
 		if len(args) == 2 {

pkg/ast/statement.go

@@ -189,6 +189,7 @@ const (
 	SLIDING_WINDOW
 	SESSION_WINDOW
 	COUNT_WINDOW
+	STATE_WINDOW
 )
 
 func (w WindowType) String() string {
@@ -211,6 +212,8 @@ func (w WindowType) String() string {
 
 type Window struct {
 	TriggerCondition Expr
+	BeginCondition   Expr
+	EmitCondition    Expr
 	WindowType       WindowType
 	Delay            *IntegerLiteral
 	Length           *IntegerLiteral