fix: fix incremental sliding delay window (#3417)

Signed-off-by: Song Gao <[email protected]>

Author: Yisaer

internal/topo/node/window_inc_agg_op.go

@@ -293,7 +293,6 @@ type SlidingWindowIncAggOp struct {
 	triggerCondition ast.Expr
 	Length           time.Duration
 	Delay            time.Duration
-	delayWindowList  []*IncAggWindow
 	currWindowList   []*IncAggWindow
 	taskCh           chan *IncAggOpTask
 }
@@ -309,7 +308,6 @@ func NewSlidingWindowIncAggOp(o *WindowIncAggOperator) *SlidingWindowIncAggOp {
 		Length:               o.windowConfig.Length,
 		Delay:                o.windowConfig.Delay,
 		currWindowList:       make([]*IncAggWindow, 0),
-		delayWindowList:      make([]*IncAggWindow, 0),
 		taskCh:               make(chan *IncAggOpTask, 1024),
 	}
 	return op
@@ -329,49 +327,31 @@ func (so *SlidingWindowIncAggOp) exec(ctx api.StreamContext, errCh chan<- error)
 			}
 			switch row := data.(type) {
 			case *xsql.Tuple:
-				so.currWindowList = gcIncAggWindow(so.currWindowList, so.Length, now)
-				if so.Delay > 0 {
-					so.appendDelayIncAggWindow(ctx, errCh, fv, row, now)
-					continue
-				}
+				so.currWindowList = gcIncAggWindow(so.currWindowList, so.Length+so.Delay, now)
 				so.appendIncAggWindow(ctx, errCh, fv, row, now)
 				if len(so.currWindowList) > 0 && so.isMatchCondition(ctx, fv, row) {
-					so.emit(ctx, errCh, so.currWindowList[0], now)
+					if so.Delay > 0 {
+						t := &IncAggOpTask{}
+						go func(task *IncAggOpTask) {
+							after := timex.After(so.Delay)
+							select {
+							case <-ctx.Done():
+								return
+							case <-after:
+								so.taskCh <- task
+							}
+						}(t)
+					} else {
+						so.emit(ctx, errCh, so.currWindowList[0], now)
+					}
 				}
 			}
-		case task := <-so.taskCh:
+		case <-so.taskCh:
 			now := timex.GetNow()
-			window := task.window
-			so.removeDelayWindow(window)
-			so.emit(ctx, errCh, window, now)
-		}
-	}
-}
-
-func (so *SlidingWindowIncAggOp) removeDelayWindow(window *IncAggWindow) {
-	if len(so.delayWindowList) == 0 {
-		return
-	}
-	if len(so.delayWindowList) == 1 {
-		if so.delayWindowList[0] == window {
-			so.delayWindowList = make([]*IncAggWindow, 0)
-		}
-		return
-	}
-	if so.delayWindowList[0] == window {
-		so.delayWindowList = so.delayWindowList[1:]
-		return
-	}
-	if so.delayWindowList[len(so.delayWindowList)-1] == window {
-		so.delayWindowList = so.delayWindowList[:len(so.delayWindowList)-1]
-		return
-	}
-	for index, w := range so.delayWindowList {
-		if w == window {
-			left := so.delayWindowList[:index]
-			right := so.delayWindowList[index+1:]
-			so.delayWindowList = append(left, right...)
-			return
+			so.currWindowList = gcIncAggWindow(so.currWindowList, so.Length+so.Delay, now)
+			if len(so.currWindowList) > 0 {
+				so.emit(ctx, errCh, so.currWindowList[0], now)
+			}
 		}
 	}
 }
@@ -384,33 +364,6 @@ func (so *SlidingWindowIncAggOp) appendIncAggWindow(ctx api.StreamContext, errCh
 	}
 }
 
-func (so *SlidingWindowIncAggOp) appendDelayIncAggWindow(ctx api.StreamContext, errCh chan<- error, fv *xsql.FunctionValuer, row *xsql.Tuple, now time.Time) {
-	name := calDimension(fv, so.Dimensions, row)
-	isMatched := so.isMatchCondition(ctx, fv, row)
-	var newDelayWindow *IncAggWindow
-	if isMatched {
-		newDelayWindow = newIncAggWindow(ctx, now)
-		so.delayWindowList = append(so.delayWindowList, newDelayWindow)
-	}
-	for _, incWindow := range so.delayWindowList {
-		if incWindow.StartTime.Add(so.Length).After(now) {
-			incAggCal(ctx, name, row, incWindow, so.aggFields)
-		}
-	}
-	if isMatched {
-		t := &IncAggOpTask{window: newDelayWindow}
-		go func(task *IncAggOpTask) {
-			after := timex.After(so.Delay)
-			select {
-			case <-ctx.Done():
-				return
-			case <-after:
-				so.taskCh <- task
-			}
-		}(t)
-	}
-}
-
 func (so *SlidingWindowIncAggOp) emit(ctx api.StreamContext, errCh chan<- error, window *IncAggWindow, now time.Time) {
 	results := &xsql.WindowTuples{
 		Content: make([]xsql.Row, 0),

internal/topo/node/window_inc_agg_op_test.go

@@ -309,7 +309,7 @@ func TestIncAggSlidingWindowDelay(t *testing.T) {
 	kv, err := store.GetKV("stream")
 	require.NoError(t, err)
 	require.NoError(t, prepareStream())
-	sql := "select count(*) from stream group by slidingWindow(ss,1,3)"
+	sql := "select count(*) from stream group by slidingWindow(ss,1,1)"
 	stmt, err := xsql.NewParser(strings.NewReader(sql)).Parse()
 	require.NoError(t, err)
 	p, err := planner.CreateLogicalPlan(stmt, &def.RuleOption{
@@ -325,7 +325,7 @@ func TestIncAggSlidingWindowDelay(t *testing.T) {
 	op, err := node.NewWindowIncAggOp("1", &node.WindowConfig{
 		Type:   incPlan.WType,
 		Length: time.Second,
-		Delay:  3 * time.Second,
+		Delay:  time.Second,
 	}, incPlan.Dimensions, incPlan.IncAggFuncs, o)
 	require.NoError(t, err)
 	require.NotNil(t, op)
@@ -338,18 +338,16 @@ func TestIncAggSlidingWindowDelay(t *testing.T) {
 	waitExecute()
 	input <- &xsql.Tuple{Message: map[string]any{"a": int64(1)}}
 	waitExecute()
-	timex.Add(300 * time.Millisecond)
+	timex.Add(500 * time.Millisecond)
 	waitExecute()
 	input <- &xsql.Tuple{Message: map[string]any{"a": int64(2)}}
 	waitExecute()
-	timex.Add(1500 * time.Millisecond)
+	timex.Add(600 * time.Millisecond)
 	waitExecute()
 	input <- &xsql.Tuple{Message: map[string]any{"a": int64(3)}}
 	waitExecute()
-	timex.Add(1300 * time.Millisecond)
+	timex.Add(2 * time.Second)
 	waitExecute()
-	timex.Add(3000 * time.Millisecond)
-
 	got := <-output
 	wt, ok := got.(*xsql.WindowTuples)
 	require.True(t, ok)
@@ -368,8 +366,8 @@ func TestIncAggSlidingWindowDelay(t *testing.T) {
 	d = wt.ToMaps()
 	require.Equal(t, []map[string]any{
 		{
-			"a":             int64(2),
-			"inc_agg_col_1": int64(1),
+			"a":             int64(3),
+			"inc_agg_col_1": int64(3),
 		},
 	}, d)
 	got = <-output
@@ -380,7 +378,7 @@ func TestIncAggSlidingWindowDelay(t *testing.T) {
 	require.Equal(t, []map[string]any{
 		{
 			"a":             int64(3),
-			"inc_agg_col_1": int64(1),
+			"inc_agg_col_1": int64(2),
 		},
 	}, d)
 	cancel()