Repository: numaproj/numaflow
Pull request #486 (2023-01-17 16:35:57)

chore: fix sliding start strategy and minor optimizations
Signed-off-by: Yashash H L <yashashhl25@gmail.com>


yhl25: > update sliding.md?

done

Patch:
diff --git a/docs/user-guide/user-defined-functions/reduce/windowing/sliding.md b/docs/user-guide/user-defined-functions/reduce/windowing/sliding.md
index f87995b1..845dd2d1 100644
--- a/docs/user-guide/user-defined-functions/reduce/windowing/sliding.md
+++ b/docs/user-guide/user-defined-functions/reduce/windowing/sliding.md
@@ -67,10 +67,9 @@ The window start time is always be left inclusive and right exclusive. That is w
 window is not considered active (it fell on the previous window, right exclusive) but `[2031-09-29T18:46:30Z, 2031-09-29T18:47:30Z)` 
 is an active (left inclusive).
 
-The first window always ends after _sliding_ seconds after the `time.now()` time. The start time itself will depend on the
-time characteristics of the pipeline (event time, system time, etc.). This means the first window has its start time 
-in the past roughly given by `time.now() - duration(length) + duration(sliding)`. So the first window starts in the past 
-and ends _sliding_duration (based on time progression in the pipeline and not the wall time) from present. It is 
+The first window always ends after the sliding seconds from the `time.Now()`, the start time of the window will be the 
+nearest integer multiple of the slide which is less than the message's event time. So the first window starts in the 
+past and ends _sliding_duration (based on time progression in the pipeline and not the wall time) from present. It is 
 important to note that regardless of the window boundary (starting in the past or ending in the future) the target element 
 set totally depends on the matching time (in case of event time, all the elements with the time that falls with in the 
 boundaries of the window, and in case of system time, all the elements that arrive from the `present` until the end of 
diff --git a/pkg/reduce/readloop/readloop.go b/pkg/reduce/readloop/readloop.go
index dd2176c1..a32f1f8c 100644
--- a/pkg/reduce/readloop/readloop.go
+++ b/pkg/reduce/readloop/readloop.go
@@ -384,9 +384,9 @@ func (rl *ReadLoop) upsertWindowsAndKeys(m *isb.ReadMessage) []window.AlignedKey
 	for _, win := range processingWindows {
 		w, isPresent := rl.windower.InsertIfNotPresent(win)
 		if !isPresent {
-			rl.log.Debugw("Creating new keyed window", zap.Any("key", w.Keys()), zap.String("msg.offset", m.ID), zap.Int64("startTime", w.StartTime().UnixMilli()), zap.Int64("endTime", w.EndTime().UnixMilli()))
+			rl.log.Debugw("Creating new keyed window", zap.String("msg.offset", m.ID), zap.Int64("startTime", w.StartTime().UnixMilli()), zap.Int64("endTime", w.EndTime().UnixMilli()))
 		} else {
-			rl.log.Debugw("Found an existing window", zap.Any("key", w.Keys()), zap.String("msg.offset", m.ID), zap.Int64("startTime", w.StartTime().UnixMilli()), zap.Int64("endTime", w.EndTime().UnixMilli()))
+			rl.log.Debugw("Found an existing window", zap.String("msg.offset", m.ID), zap.Int64("startTime", w.StartTime().UnixMilli()), zap.Int64("endTime", w.EndTime().UnixMilli()))
 		}
 		// track the key to window relationship
 		w.AddKey(m.Key)
diff --git a/pkg/window/strategy/sliding/sliding.go b/pkg/window/strategy/sliding/sliding.go
index a5bdc525..ff4c76ec 100644
--- a/pkg/window/strategy/sliding/sliding.go
+++ b/pkg/window/strategy/sliding/sliding.go
@@ -65,47 +65,32 @@ func NewSliding(length time.Duration, slide time.Duration) *Sliding {
 
 // AssignWindow returns a set of windows that contain the element based on event time
 func (s *Sliding) AssignWindow(eventTime time.Time) []window.AlignedKeyedWindower {
-	// start time of the window in to which this element certainly belongs.
-	startTime := eventTime.Truncate(s.Length)
-	// end time of the window in to which this element certainly belongs.
-	endTime := startTime.Add(s.Length)
-
-	// we have to find the boundaries of the sliding windows that are possible
-	// first lets consider end time as fixed, find the min start and end times
-	minEndTime := startTime.Add(s.Length % s.Slide)
-	minStartTime := minEndTime.Add(-s.Length)
-
-	// lets consider start time as fixed and find the max start and end times.
-	maxStartTime := endTime.Add(-(s.Length % s.Slide))
-	maxEndTime := maxStartTime.Add(s.Length)
-
-	// now all the windows should fall in between maxend and minend times.
-	// one could consider min start and max start times as well.
-	wCount := int((maxEndTime.Sub(minEndTime)) / s.Slide)
 	windows := make([]window.AlignedKeyedWindower, 0)
 
-	for i := 0; i < wCount; i++ {
-		// we make the windows left aligned since the original truncation operation
-		// is left aligned.
-		st := minStartTime.Add(time.Duration(i) * s.Slide)
-		et := st.Add(s.Length)
-
-		// since there is overlap at the boundaries
-		// we attribute the element to the window to the right (higher)
-		// of the boundary
-		// left exclusive and right inclusive
-		// so given windows 500-600 and 600-700 and the event time is 600
-		// we will add the element to 600-700 window and not to the 500-600 window.
-		if eventTime.Before(st) || !eventTime.Before(et) {
-			continue
-		}
-
-		akw := keyed.NewKeyedWindow(st, et)
+	// use the highest integer multiple of slide length which is less than the eventTime
+	// as the start time for the window. For example if the eventTime is 810 and slide
+	// length is 70, use 770 as the startTime of the window. In that way we can be guarantee
+	// consistency while assigning the messages to the windows.
+	startTime := time.UnixMilli((eventTime.UnixMilli() / s.Slide.Milliseconds()) * s.Slide.Milliseconds())
+	endTime := startTime.Add(s.Length)
 
-		windows = append(windows, akw)
+	// startTime and endTime will be the largest timestamp window for the given eventTime,
+	// using that we can create other windows by subtracting the slide length
+
+	// since there is overlap at the boundaries
+	// we attribute the element to the window to the right (higher)
+	// of the boundary
+	// left exclusive and right inclusive
+	// so given windows 500-600 and 600-700 and the event time is 600
+	// we will add the element to 600-700 window and not to the 500-600 window.
+	for !startTime.After(eventTime) && endTime.After(eventTime) {
+		windows = append(windows, keyed.NewKeyedWindow(startTime, endTime))
+		startTime = startTime.Add(-s.Slide)
+		endTime = endTime.Add(-s.Slide)
 	}
 
 	return windows
+
 }
 
 // InsertIfNotPresent inserts a window to the list of active windows if not present and returns the window


Test patch:
diff --git a/pkg/window/strategy/sliding/sliding_test.go b/pkg/window/strategy/sliding/sliding_test.go
index f0395684..50287919 100644
--- a/pkg/window/strategy/sliding/sliding_test.go
+++ b/pkg/window/strategy/sliding/sliding_test.go
@@ -21,10 +21,9 @@ import (
 	"testing"
 	"time"
 
-	"github.com/stretchr/testify/assert"
-
 	"github.com/numaproj/numaflow/pkg/window"
 	"github.com/numaproj/numaflow/pkg/window/keyed"
+	"github.com/stretchr/testify/assert"
 )
 
 // TestSliding_AssignWindow tests the assignment of element to a set of windows
@@ -44,9 +43,9 @@ func TestSliding_AssignWindow(t *testing.T) {
 			slide:     20 * time.Second,
 			eventTime: baseTime.Add(10 * time.Second),
 			expected: []window.AlignedKeyedWindower{
-				keyed.NewKeyedWindow(time.Unix(560, 0), time.Unix(620, 0)),
-				keyed.NewKeyedWindow(time.Unix(580, 0), time.Unix(640, 0)),
 				keyed.NewKeyedWindow(time.Unix(600, 0), time.Unix(660, 0)),
+				keyed.NewKeyedWindow(time.Unix(580, 0), time.Unix(640, 0)),
+				keyed.NewKeyedWindow(time.Unix(560, 0), time.Unix(620, 0)),
 			},
 		},
 		{
@@ -55,8 +54,8 @@ func TestSliding_AssignWindow(t *testing.T) {
 			slide:     40 * time.Second,
 			eventTime: baseTime.Add(10 * time.Second),
 			expected: []window.AlignedKeyedWindower{
-				keyed.NewKeyedWindow(time.Unix(560, 0), time.Unix(620, 0)),
 				keyed.NewKeyedWindow(time.Unix(600, 0), time.Unix(660, 0)),
+				keyed.NewKeyedWindow(time.Unix(560, 0), time.Unix(620, 0)),
 			},
 		},
 		{
@@ -65,8 +64,7 @@ func TestSliding_AssignWindow(t *testing.T) {
 			slide:     41 * time.Second,
 			eventTime: baseTime.Add(10 * time.Second),
 			expected: []window.AlignedKeyedWindower{
-				keyed.NewKeyedWindow(time.Unix(559, 0), time.Unix(619, 0)),
-				keyed.NewKeyedWindow(time.Unix(600, 0), time.Unix(660, 0)),
+				keyed.NewKeyedWindow(time.Unix(574, 0), time.Unix(634, 0)),
 			},
 		},
 		{
@@ -75,9 +73,9 @@ func TestSliding_AssignWindow(t *testing.T) {
 			slide:     20 * time.Second,
 			eventTime: baseTime,
 			expected: []window.AlignedKeyedWindower{
-				keyed.NewKeyedWindow(time.Unix(560, 0), time.Unix(620, 0)),
-				keyed.NewKeyedWindow(time.Unix(580, 0), time.Unix(640, 0)),
 				keyed.NewKeyedWindow(time.Unix(600, 0), time.Unix(660, 0)),
+				keyed.NewKeyedWindow(time.Unix(580, 0), time.Unix(640, 0)),
+				keyed.NewKeyedWindow(time.Unix(560, 0), time.Unix(620, 0)),
 			},
 		},
 		{
@@ -86,9 +84,9 @@ func TestSliding_AssignWindow(t *testing.T) {
 			slide:     20 * time.Second,
 			eventTime: baseTime.Add(time.Minute),
 			expected: []window.AlignedKeyedWindower{
-				keyed.NewKeyedWindow(time.Unix(620, 0), time.Unix(680, 0)),
-				keyed.NewKeyedWindow(time.Unix(640, 0), time.Unix(700, 0)),
 				keyed.NewKeyedWindow(time.Unix(660, 0), time.Unix(720, 0)),
+				keyed.NewKeyedWindow(time.Unix(640, 0), time.Unix(700, 0)),
+				keyed.NewKeyedWindow(time.Unix(620, 0), time.Unix(680, 0)),
 			},
 		},
 		{
@@ -97,9 +95,9 @@ func TestSliding_AssignWindow(t *testing.T) {
 			slide:     20 * time.Second,
 			eventTime: baseTime.Add(time.Nanosecond),
 			expected: []window.AlignedKeyedWindower{
-				keyed.NewKeyedWindow(time.Unix(560, 0), time.Unix(620, 0)),
-				keyed.NewKeyedWindow(time.Unix(580, 0), time.Unix(640, 0)),
 				keyed.NewKeyedWindow(time.Unix(600, 0), time.Unix(660, 0)),
+				keyed.NewKeyedWindow(time.Unix(580, 0), time.Unix(640, 0)),
+				keyed.NewKeyedWindow(time.Unix(560, 0), time.Unix(620, 0)),
 			},
 		},
 		{
@@ -108,9 +106,9 @@ func TestSliding_AssignWindow(t *testing.T) {
 			slide:     20 * time.Second,
 			eventTime: baseTime.Add(-time.Nanosecond),
 			expected: []window.AlignedKeyedWindower{
-				keyed.NewKeyedWindow(time.Unix(540, 0), time.Unix(600, 0)),
-				keyed.NewKeyedWindow(time.Unix(560, 0), time.Unix(620, 0)),
 				keyed.NewKeyedWindow(time.Unix(580, 0), time.Unix(640, 0)),
+				keyed.NewKeyedWindow(time.Unix(560, 0), time.Unix(620, 0)),
+				keyed.NewKeyedWindow(time.Unix(540, 0), time.Unix(600, 0)),
 			},
 		},
 		{
@@ -119,9 +117,9 @@ func TestSliding_AssignWindow(t *testing.T) {
 			slide:     20 * time.Second,
 			eventTime: baseTime.Add(20 * time.Second),
 			expected: []window.AlignedKeyedWindower{
-				keyed.NewKeyedWindow(time.Unix(580, 0), time.Unix(640, 0)),
-				keyed.NewKeyedWindow(time.Unix(600, 0), time.Unix(660, 0)),
 				keyed.NewKeyedWindow(time.Unix(620, 0), time.Unix(680, 0)),
+				keyed.NewKeyedWindow(time.Unix(600, 0), time.Unix(660, 0)),
+				keyed.NewKeyedWindow(time.Unix(580, 0), time.Unix(640, 0)),
 			},
 		},
 		{
@@ -130,9 +128,42 @@ func TestSliding_AssignWindow(t *testing.T) {
 			slide:     20 * time.Second,
 			eventTime: baseTime.Add(21 * time.Second),
 			expected: []window.AlignedKeyedWindower{
-				keyed.NewKeyedWindow(time.Unix(580, 0), time.Unix(640, 0)),
-				keyed.NewKeyedWindow(time.Unix(600, 0), time.Unix(660, 0)),
 				keyed.NewKeyedWindow(time.Unix(620, 0), time.Unix(680, 0)),
+				keyed.NewKeyedWindow(time.Unix(600, 0), time.Unix(660, 0)),
+				keyed.NewKeyedWindow(time.Unix(580, 0), time.Unix(640, 0)),
+			},
+		},
+		{
+			name:      "length not divisible by slide test 1",
+			length:    time.Second * 600,
+			slide:     70 * time.Second,
+			eventTime: baseTime.Add(210 * time.Second), // 810
+			expected: []window.AlignedKeyedWindower{
+				keyed.NewKeyedWindow(time.Unix(770, 0), time.Unix(1370, 0)),
+				keyed.NewKeyedWindow(time.Unix(700, 0), time.Unix(1300, 0)),
+				keyed.NewKeyedWindow(time.Unix(630, 0), time.Unix(1230, 0)),
+				keyed.NewKeyedWindow(time.Unix(560, 0), time.Unix(1160, 0)),
+				keyed.NewKeyedWindow(time.Unix(490, 0), time.Unix(1090, 0)),
+				keyed.NewKeyedWindow(time.Unix(420, 0), time.Unix(1020, 0)),
+				keyed.NewKeyedWindow(time.Unix(350, 0), time.Unix(950, 0)),
+				keyed.NewKeyedWindow(time.Unix(280, 0), time.Unix(880, 0)),
+			},
+		},
+		{
+			name:      "length not divisible by slide test 2",
+			length:    time.Second * 600,
+			slide:     70 * time.Second,
+			eventTime: baseTime.Add(610 * time.Second), // 1210
+			expected: []window.AlignedKeyedWindower{
+				keyed.NewKeyedWindow(time.Unix(1190, 0), time.Unix(1790, 0)),
+				keyed.NewKeyedWindow(time.Unix(1120, 0), time.Unix(1720, 0)),
+				keyed.NewKeyedWindow(time.Unix(1050, 0), time.Unix(1650, 0)),
+				keyed.NewKeyedWindow(time.Unix(980, 0), time.Unix(1580, 0)),
+				keyed.NewKeyedWindow(time.Unix(910, 0), time.Unix(1510, 0)),
+				keyed.NewKeyedWindow(time.Unix(840, 0), time.Unix(1440, 0)),
+				keyed.NewKeyedWindow(time.Unix(770, 0), time.Unix(1370, 0)),
+				keyed.NewKeyedWindow(time.Unix(700, 0), time.Unix(1300, 0)),
+				keyed.NewKeyedWindow(time.Unix(630, 0), time.Unix(1230, 0)),
 			},
 		},
 	}