Repository: numaproj/numaflow
Pull request #409 (2022-12-06 14:02:45)

chore: Windower interface refactor
Signed-off-by: Yashash H L <yashashhl25@gmail.com>

yhl25: @ashwinidulams @vigith  I was also thinking should we merge the insert and get method to one method like (InsertIfNotPresent) to avoid iterating over the list twice
vigith: > @ashwinidulams @vigith I was also thinking should we merge the insert and get method to one method like (InsertIfNotPresent) to avoid iterating over the list twice

That makes sense; let's try that?
yhl25: > > @ashwinidulams @vigith I was also thinking should we merge the insert and get method to one method like (InsertIfNotPresent) to avoid iterating over the list twice
> 
> That makes sense; let's try that?

done

Patch:
diff --git a/pkg/reduce/readloop/readloop.go b/pkg/reduce/readloop/readloop.go
index c780a011..16401483 100644
--- a/pkg/reduce/readloop/readloop.go
+++ b/pkg/reduce/readloop/readloop.go
@@ -103,12 +103,13 @@ func (rl *ReadLoop) Startup(ctx context.Context) error {
 		// crosses the window.
 
 		alignedKeyedWindow := keyed.NewKeyedWindow(p.Start, p.End)
+
+		// insert the window to the list of active windows, since the active window list is in-memory
+		keyedWindow, _ := rl.windower.InsertIfNotPresent(alignedKeyedWindow)
+
 		// add key to the window, so that when a new message with the watermark greater than
 		// the window end time comes, key will not be lost and the windows will be closed as expected
-		alignedKeyedWindow.AddKey(p.Key)
-
-		// These windows have to be recreated as they are completely in-memory
-		rl.windower.CreateWindow(alignedKeyedWindow)
+		keyedWindow.AddKey(p.Key)
 
 		// create and invoke process and forward for the partition
 		rl.associatePBQAndPnF(ctx, p)
@@ -328,9 +329,8 @@ func (rl *ReadLoop) upsertWindowsAndKeys(m *isb.ReadMessage) []window.AlignedKey
 	processingWindows := rl.windower.AssignWindow(m.EventTime)
 	var kWindows []window.AlignedKeyedWindower
 	for _, win := range processingWindows {
-		w := rl.windower.GetWindow(win)
-		if w == nil {
-			w = rl.windower.CreateWindow(win)
+		w, isNew := rl.windower.InsertIfNotPresent(win)
+		if isNew {
 			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()))
 		} 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()))
diff --git a/pkg/window/strategy/fixed/fixed.go b/pkg/window/strategy/fixed/fixed.go
index 68490c7a..b6c575ec 100644
--- a/pkg/window/strategy/fixed/fixed.go
+++ b/pkg/window/strategy/fixed/fixed.go
@@ -77,74 +77,50 @@ func (f *Fixed) AssignWindow(eventTime time.Time) []window.AlignedKeyedWindower
 	}
 }
 
-// CreateWindow adds a window for a given interval window
-func (f *Fixed) CreateWindow(kw window.AlignedKeyedWindower) window.AlignedKeyedWindower {
+// InsertIfNotPresent inserts a window to the list of active windows if not present and returns the window
+func (f *Fixed) InsertIfNotPresent(kw window.AlignedKeyedWindower) (aw window.AlignedKeyedWindower, isPresent bool) {
 	f.lock.Lock()
 	defer f.lock.Unlock()
 
 	// this could be the first window
 	if f.entries.Len() == 0 {
 		f.entries.PushFront(kw)
-		return kw
+		return kw, false
 	}
 
 	earliestWindow := f.entries.Front().Value.(*keyed.AlignedKeyedWindow)
 	recentWindow := f.entries.Back().Value.(*keyed.AlignedKeyedWindow)
 
-	// late arrival
-	if !earliestWindow.StartTime().Before(kw.EndTime()) {
+	// if there is only one window
+	if earliestWindow.StartTime().Equal(kw.StartTime()) && earliestWindow.EndTime().Equal(kw.EndTime()) {
+		aw = earliestWindow
+		isPresent = true
+	} else if !earliestWindow.StartTime().Before(kw.EndTime()) {
+		// late arrival
 		f.entries.PushFront(kw)
+		aw = kw
 	} else if !recentWindow.EndTime().After(kw.StartTime()) {
 		// early arrival
 		f.entries.PushBack(kw)
+		aw = kw
 	} else {
 		// a window in the middle
-		for e := f.entries.Back(); e != nil; e = e.Prev() {
+		for e := f.entries.Back(); e.Prev() != nil; e = e.Prev() {
 			win := e.Value.(*keyed.AlignedKeyedWindow)
-			if !win.StartTime().Before(kw.EndTime()) {
+			prevWin := e.Prev().Value.(*keyed.AlignedKeyedWindow)
+			if win.StartTime().Equal(kw.StartTime()) && win.EndTime().Equal(kw.EndTime()) {
+				aw = win
+				isPresent = true
+				break
+			}
+			if !win.StartTime().Before(kw.EndTime()) && !prevWin.EndTime().After(kw.StartTime()) {
 				f.entries.InsertBefore(kw, e)
+				aw = kw
 				break
 			}
 		}
 	}
-	return kw
-}
-
-// GetWindow returns an existing window for the given interval
-func (f *Fixed) GetWindow(kw window.AlignedKeyedWindower) window.AlignedKeyedWindower {
-	f.lock.RLock()
-	defer f.lock.RUnlock()
-
-	if f.entries.Len() == 0 {
-		return nil
-	}
-
-	// are we looking for a window that is later than the current latest?
-	latest := f.entries.Back()
-	lkw := latest.Value.(*keyed.AlignedKeyedWindow)
-	if !lkw.EndTime().After(kw.StartTime()) {
-		return nil
-	}
-
-	// are we looking for a window that is earlier than the current earliest?
-	earliest := f.entries.Front()
-	ekw := earliest.Value.(*keyed.AlignedKeyedWindow)
-	if !ekw.StartTime().Before(kw.EndTime()) {
-		return nil
-	}
-
-	// check if we already have a window
-	for e := f.entries.Back(); e != nil; e = e.Prev() {
-		win := e.Value.(*keyed.AlignedKeyedWindow)
-		if win.StartTime().Equal(kw.StartTime()) && win.EndTime().Equal(kw.EndTime()) {
-			return win
-		} else if win.StartTime().Before(kw.EndTime()) {
-			// we have moved past the range that we are looking for
-			// so, we can bail out early.
-			break
-		}
-	}
-	return nil
+	return
 }
 
 // RemoveWindows returns an array of keyed windows that are before the current watermark.
diff --git a/pkg/window/windower.go b/pkg/window/windower.go
index b6d17959..b2a57801 100644
--- a/pkg/window/windower.go
+++ b/pkg/window/windower.go
@@ -42,10 +42,9 @@ type AlignedKeyedWindower interface {
 type Windower interface {
 	// AssignWindow assigns the event to the window based on give window configuration.
 	AssignWindow(eventTime time.Time) []AlignedKeyedWindower
-	// CreateWindow creates a window for a supplied interval
-	CreateWindow(aw AlignedKeyedWindower) AlignedKeyedWindower
-	// GetWindow returns a keyed window for a supplied interval
-	GetWindow(aw AlignedKeyedWindower) AlignedKeyedWindower
+	// InsertIfNotPresent inserts window to the list of active windows if not present
+	// if present it will return the window
+	InsertIfNotPresent(aw AlignedKeyedWindower) (AlignedKeyedWindower, bool)
 	// RemoveWindows returns list of window(s) that can be closed
 	RemoveWindows(time time.Time) []AlignedKeyedWindower
 }


Test patch:
diff --git a/pkg/reduce/readloop/readloop_test.go b/pkg/reduce/readloop/readloop_test.go
index 5679bd07..9f453c4c 100644
--- a/pkg/reduce/readloop/readloop_test.go
+++ b/pkg/reduce/readloop/readloop_test.go
@@ -69,6 +69,11 @@ func TestReadLoop_Startup(t *testing.T) {
 			End:   time.Unix(120, 0),
 			Key:   "even",
 		},
+		{
+			Start: time.Unix(60, 0),
+			End:   time.Unix(120, 0),
+			Key:   "odd",
+		},
 		{
 			Start: time.Unix(120, 0),
 			End:   time.Unix(180, 0),
@@ -104,7 +109,7 @@ func TestReadLoop_Startup(t *testing.T) {
 
 	pManager, _ := pbq.NewManager(ctx, memStoreProvider, pbq.WithChannelBufferSize(10))
 
-	to1 := simplebuffer.NewInMemoryBuffer("reduce-buffer", 3)
+	to1 := simplebuffer.NewInMemoryBuffer("reduce-buffer", 4)
 	toSteps := map[string]isb.BufferWriter{
 		"reduce-buffer": to1,
 	}
@@ -148,25 +153,29 @@ func TestReadLoop_Startup(t *testing.T) {
 		}
 	}
 
-	msgs, readErr := to1.Read(ctx, 3)
+	msgs, readErr := to1.Read(ctx, 4)
 	assert.Nil(t, readErr)
-	assert.Len(t, msgs, 3)
+	assert.Len(t, msgs, 4)
 
 	// since we have 3 partitions we should have 3 different outputs
 	var readMessagePayload1 PayloadForTest
 	var readMessagePayload2 PayloadForTest
 	var readMessagePayload3 PayloadForTest
+	var readMessagePayload4 PayloadForTest
 	_ = json.Unmarshal(msgs[0].Payload, &readMessagePayload1)
 	_ = json.Unmarshal(msgs[1].Payload, &readMessagePayload2)
 	_ = json.Unmarshal(msgs[2].Payload, &readMessagePayload3)
+	_ = json.Unmarshal(msgs[3].Payload, &readMessagePayload4)
 	// since we had 10 messages in the store with value 2 and 3
 	// the expected value is 20 and 30, since the reduce operation is sum
-	assert.Contains(t, []int{20, 30, 20}, readMessagePayload1.Value)
-	assert.Contains(t, []int{20, 30, 20}, readMessagePayload2.Value)
-	assert.Contains(t, []int{20, 30, 20}, readMessagePayload3.Value)
+	assert.Contains(t, []int{20, 30, 20, 30}, readMessagePayload1.Value)
+	assert.Contains(t, []int{20, 30, 20, 30}, readMessagePayload2.Value)
+	assert.Contains(t, []int{20, 30, 20, 30}, readMessagePayload3.Value)
+	assert.Contains(t, []int{20, 30, 20, 30}, readMessagePayload4.Value)
 	assert.Equal(t, "sum", readMessagePayload1.Key)
 	assert.Equal(t, "sum", readMessagePayload2.Key)
 	assert.Equal(t, "sum", readMessagePayload3.Key)
+	assert.Equal(t, "sum", readMessagePayload4.Key)
 
 }
 
diff --git a/pkg/window/strategy/fixed/fixed_test.go b/pkg/window/strategy/fixed/fixed_test.go
index 4446c175..719f2078 100644
--- a/pkg/window/strategy/fixed/fixed_test.go
+++ b/pkg/window/strategy/fixed/fixed_test.go
@@ -94,13 +94,14 @@ func TestFixed_AssignWindow(t *testing.T) {
 
 // TestAligned_CreateWindow tests the insertion of a new keyed window for a given interval
 // It tests early, late and existing window scenarios.
-func TestAligned_CreateWindow(t *testing.T) {
+func TestAligned_InsertIfNotPresent(t *testing.T) {
 	windows := NewFixed(60 * time.Second)
 	tests := []struct {
 		name            string
 		given           []*keyed.AlignedKeyedWindow
 		input           *keyed.AlignedKeyedWindow
 		expectedWindows []*keyed.AlignedKeyedWindow
+		isPresent       bool
 	}{
 		{
 			name:  "FirstWindow",
@@ -115,6 +116,7 @@ func TestAligned_CreateWindow(t *testing.T) {
 					End:   time.Unix(60, 0),
 				},
 			},
+			isPresent: false,
 		},
 		{
 			name: "late_window",
@@ -138,6 +140,7 @@ func TestAligned_CreateWindow(t *testing.T) {
 					End:   time.Unix(180, 0),
 				},
 			},
+			isPresent: false,
 		},
 		{
 			name: "early_window",
@@ -161,6 +164,7 @@ func TestAligned_CreateWindow(t *testing.T) {
 					End:   time.Unix(300, 0),
 				},
 			},
+			isPresent: false,
 		},
 		{
 			name: "insert_middle",
@@ -174,10 +178,7 @@ func TestAligned_CreateWindow(t *testing.T) {
 					End:   time.Unix(300, 0),
 				},
 			},
-			input: &keyed.AlignedKeyedWindow{
-				Start: time.Unix(180, 0),
-				End:   time.Unix(240, 0),
-			},
+			input: keyed.NewKeyedWindow(time.Unix(180, 0), time.Unix(240, 0)),
 			expectedWindows: []*keyed.AlignedKeyedWindow{
 				{
 					Start: time.Unix(120, 0),
@@ -192,174 +193,57 @@ func TestAligned_CreateWindow(t *testing.T) {
 					End:   time.Unix(300, 0),
 				},
 			},
-		},
-	}
-
-	for _, tt := range tests {
-		t.Run(tt.name, func(t *testing.T) {
-			windows := windows.(*Fixed)
-			setup(windows, tt.given)
-			ret := windows.CreateWindow(tt.input)
-			assert.Equal(t, tt.input.Start, ret.StartTime())
-			assert.Equal(t, tt.input.End, ret.EndTime())
-			assert.Equal(t, len(tt.expectedWindows), windows.entries.Len())
-			node := windows.entries.Front()
-			for _, kw := range tt.expectedWindows {
-				assert.Equal(t, kw.StartTime(), node.Value.(*keyed.AlignedKeyedWindow).StartTime())
-				assert.Equal(t, kw.EndTime(), node.Value.(*keyed.AlignedKeyedWindow).EndTime())
-				node = node.Next()
-			}
-		})
-	}
-}
-
-func TestAligned_GetWindow(t *testing.T) {
-	windows := NewFixed(60 * time.Second)
-	tests := []struct {
-		name           string
-		given          []*keyed.AlignedKeyedWindow
-		input          *keyed.AlignedKeyedWindow
-		expectedWindow *keyed.AlignedKeyedWindow
-	}{
-		{
-			name:  "non_existing",
-			given: []*keyed.AlignedKeyedWindow{},
-			input: &keyed.AlignedKeyedWindow{
-				Start: time.Unix(0, 0),
-				End:   time.Unix(60, 0),
-			},
-			expectedWindow: nil,
-		},
-		{
-			name: "non_existing_before_earlier",
-			given: []*keyed.AlignedKeyedWindow{
-				{
-					Start: time.Unix(60, 0),
-					End:   time.Unix(120, 0),
-				},
-				{
-					Start: time.Unix(120, 0),
-					End:   time.Unix(180, 0),
-				},
-			},
-			input: &keyed.AlignedKeyedWindow{
-				Start: time.Unix(0, 0),
-				End:   time.Unix(60, 0),
-			},
-			expectedWindow: nil,
+			isPresent: false,
 		},
 		{
-			name: "non_existing_after_recent",
+			name: "already_present",
 			given: []*keyed.AlignedKeyedWindow{
-				{
-					Start: time.Unix(60, 0),
-					End:   time.Unix(120, 0),
-				},
 				{
 					Start: time.Unix(120, 0),
 					End:   time.Unix(180, 0),
 				},
-			},
-			input: &keyed.AlignedKeyedWindow{
-				Start: time.Unix(180, 0),
-				End:   time.Unix(240, 0),
-			},
-			expectedWindow: nil,
-		},
-		{
-			name: "non_existing_middle",
-			given: []*keyed.AlignedKeyedWindow{
-				{
-					Start: time.Unix(60, 0),
-					End:   time.Unix(120, 0),
-				},
 				{
 					Start: time.Unix(180, 0),
 					End:   time.Unix(240, 0),
 				},
-			},
-			input: &keyed.AlignedKeyedWindow{
-				Start: time.Unix(120, 0),
-				End:   time.Unix(180, 0),
-			},
-			expectedWindow: nil,
-		},
-		{
-			name: "existing",
-			given: []*keyed.AlignedKeyedWindow{
-				{
-					Start: time.Unix(60, 0),
-					End:   time.Unix(120, 0),
-				},
-				{
-					Start: time.Unix(120, 0),
-					End:   time.Unix(180, 0),
-				},
-			},
-			input: &keyed.AlignedKeyedWindow{
-				Start: time.Unix(60, 0),
-				End:   time.Unix(120, 0),
-			},
-			expectedWindow: &keyed.AlignedKeyedWindow{
-				Start: time.Unix(60, 0),
-				End:   time.Unix(120, 0),
-			},
-		},
-		{
-			name: "first_window",
-			given: []*keyed.AlignedKeyedWindow{
-				{
-					Start: time.Unix(120, 0),
-					End:   time.Unix(180, 0),
-				},
 				{
 					Start: time.Unix(240, 0),
 					End:   time.Unix(300, 0),
 				},
 			},
-			input: &keyed.AlignedKeyedWindow{
-				Start: time.Unix(120, 0),
-				End:   time.Unix(180, 0),
-			},
-			expectedWindow: &keyed.AlignedKeyedWindow{
-				Start: time.Unix(120, 0),
-				End:   time.Unix(180, 0),
-			},
-		},
-		{
-			name: "last_window",
-			given: []*keyed.AlignedKeyedWindow{
+			input: keyed.NewKeyedWindow(time.Unix(180, 0), time.Unix(240, 0)),
+			expectedWindows: []*keyed.AlignedKeyedWindow{
 				{
 					Start: time.Unix(120, 0),
 					End:   time.Unix(180, 0),
 				},
+				{
+					Start: time.Unix(180, 0),
+					End:   time.Unix(240, 0),
+				},
 				{
 					Start: time.Unix(240, 0),
 					End:   time.Unix(300, 0),
 				},
 			},
-			input: &keyed.AlignedKeyedWindow{
-				Start: time.Unix(240, 0),
-				End:   time.Unix(300, 0),
-			},
-			expectedWindow: &keyed.AlignedKeyedWindow{
-				Start: time.Unix(240, 0),
-				End:   time.Unix(300, 0),
-			},
+			isPresent: true,
 		},
 	}
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			setup(windows.(*Fixed), tt.given)
-			ret := windows.GetWindow(tt.input)
-			if tt.expectedWindow == nil {
-				assert.Nil(t, ret)
-			} else {
-				assert.Equal(t, tt.input.StartTime(), ret.StartTime())
-				assert.Equal(t, tt.input.EndTime(), ret.EndTime())
-				assert.Equal(t, tt.expectedWindow.StartTime(), ret.StartTime())
-				assert.Equal(t, tt.expectedWindow.EndTime(), ret.EndTime())
+			windows := windows.(*Fixed)
+			setup(windows, tt.given)
+			ret, isPresent := windows.InsertIfNotPresent(tt.input)
+			assert.Equal(t, tt.isPresent, isPresent)
+			assert.Equal(t, tt.input.Start, ret.StartTime())
+			assert.Equal(t, tt.input.End, ret.EndTime())
+			assert.Equal(t, len(tt.expectedWindows), windows.entries.Len())
+			node := windows.entries.Front()
+			for _, kw := range tt.expectedWindows {
+				assert.Equal(t, kw.StartTime(), node.Value.(*keyed.AlignedKeyedWindow).StartTime())
+				assert.Equal(t, kw.EndTime(), node.Value.(*keyed.AlignedKeyedWindow).EndTime())
+				node = node.Next()
 			}
 		})
 	}