Repository: numaproj/numaflow
Pull request #347 (2022-11-14 17:40:32)

chore: minor window refactor
Signed-off-by: Vigith Maurice <vigith@gmail.com>




Patch:
diff --git a/pkg/reduce/readloop/readloop.go b/pkg/reduce/readloop/readloop.go
index 4e4d0ed9..86cd008b 100644
--- a/pkg/reduce/readloop/readloop.go
+++ b/pkg/reduce/readloop/readloop.go
@@ -99,7 +99,7 @@ func (rl *ReadLoop) Startup(ctx context.Context) {
 		// so that the window can be closed when the watermark
 		// crosses the window.
 		id := p.PartitionID
-		intervalWindow := &keyed.KeyedWindow{
+		intervalWindow := &keyed.AlignedKeyedWindow{
 			Start: id.Start,
 			End:   id.End,
 		}
@@ -229,15 +229,15 @@ func (rl *ReadLoop) ShutDown(ctx context.Context) {
 
 // upsertWindowsAndKeys will create or assigns (if already present) a window to the message. It is an upsert operation
 // because windows are created out of order, but they will be closed in-order.
-func (rl *ReadLoop) upsertWindowsAndKeys(m *isb.ReadMessage) []window.AlignedWindow {
+func (rl *ReadLoop) upsertWindowsAndKeys(m *isb.ReadMessage) []window.AlignedKeyedWindower {
 	// drop the late messages
 	if m.IsLate {
 		rl.log.Warnw("Dropping the late message", zap.Time("eventTime", m.EventTime), zap.Time("watermark", m.Watermark))
-		return []window.AlignedWindow{}
+		return []window.AlignedKeyedWindower{}
 	}
 
 	processingWindows := rl.windower.AssignWindow(m.EventTime)
-	var kWindows []window.AlignedWindow
+	var kWindows []window.AlignedKeyedWindower
 	for _, win := range processingWindows {
 		w := rl.windower.GetWindow(win)
 		if w == nil {
diff --git a/pkg/watermark/ot/ot.go b/pkg/watermark/ot/ot.go
index 65426886..c853ba04 100644
--- a/pkg/watermark/ot/ot.go
+++ b/pkg/watermark/ot/ot.go
@@ -1,3 +1,19 @@
+/*
+Copyright 2022 The Numaproj Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
 // Package ot represents the offset-timeline pair and its corresponding encoder and decoder.
 package ot
 
diff --git a/pkg/window/keyed/keyed.go b/pkg/window/keyed/keyed.go
index f9c8e31d..ee668cb7 100644
--- a/pkg/window/keyed/keyed.go
+++ b/pkg/window/keyed/keyed.go
@@ -23,11 +23,12 @@ import (
 	"time"
 
 	"github.com/numaproj/numaflow/pkg/pbq/partition"
+	"github.com/numaproj/numaflow/pkg/window"
 )
 
-// KeyedWindow maintains association between keys and a window.
+// AlignedKeyedWindow maintains association between keys and a window.
 // In a keyed stream, we need to close all the partitions when the watermark is past the window.
-type KeyedWindow struct {
+type AlignedKeyedWindow struct {
 	// Start start time of the window
 	Start time.Time
 	// End end time of the window
@@ -37,9 +38,11 @@ type KeyedWindow struct {
 	lock sync.RWMutex
 }
 
+var _ window.AlignedKeyedWindower = (*AlignedKeyedWindow)(nil)
+
 // NewKeyedWindow creates a new keyed window
-func NewKeyedWindow(start time.Time, end time.Time) *KeyedWindow {
-	kw := &KeyedWindow{
+func NewKeyedWindow(start time.Time, end time.Time) *AlignedKeyedWindow {
+	kw := &AlignedKeyedWindow{
 		Start: start,
 		End:   end,
 		keys:  make(map[string]struct{}),
@@ -49,17 +52,17 @@ func NewKeyedWindow(start time.Time, end time.Time) *KeyedWindow {
 }
 
 // StartTime returns start of the window.
-func (kw *KeyedWindow) StartTime() time.Time {
+func (kw *AlignedKeyedWindow) StartTime() time.Time {
 	return kw.Start
 }
 
 // EndTime returns end of the window.
-func (kw *KeyedWindow) EndTime() time.Time {
+func (kw *AlignedKeyedWindow) EndTime() time.Time {
 	return kw.End
 }
 
 // AddKey adds a key to an existing window
-func (kw *KeyedWindow) AddKey(key string) {
+func (kw *AlignedKeyedWindow) AddKey(key string) {
 	kw.lock.Lock()
 	defer kw.lock.Unlock()
 	if _, ok := kw.keys[key]; !ok {
@@ -68,7 +71,7 @@ func (kw *KeyedWindow) AddKey(key string) {
 }
 
 // Partitions returns an array of partitions for a window
-func (kw *KeyedWindow) Partitions() []partition.ID {
+func (kw *AlignedKeyedWindow) Partitions() []partition.ID {
 	kw.lock.RLock()
 	defer kw.lock.RUnlock()
 
@@ -82,7 +85,7 @@ func (kw *KeyedWindow) Partitions() []partition.ID {
 	return partitions
 }
 
-func (kw *KeyedWindow) Keys() []string {
+func (kw *AlignedKeyedWindow) Keys() []string {
 	kw.lock.RLock()
 	defer kw.lock.RUnlock()
 
diff --git a/pkg/window/strategy/fixed/fixed.go b/pkg/window/strategy/fixed/fixed.go
index f9b0260a..68490c7a 100644
--- a/pkg/window/strategy/fixed/fixed.go
+++ b/pkg/window/strategy/fixed/fixed.go
@@ -64,7 +64,7 @@ func NewFixed(length time.Duration) window.Windower {
 }
 
 // AssignWindow assigns a window for the given eventTime.
-func (f *Fixed) AssignWindow(eventTime time.Time) []window.AlignedWindow {
+func (f *Fixed) AssignWindow(eventTime time.Time) []window.AlignedKeyedWindower {
 	start := eventTime.Truncate(f.Length)
 	end := start.Add(f.Length)
 
@@ -72,13 +72,13 @@ func (f *Fixed) AssignWindow(eventTime time.Time) []window.AlignedWindow {
 	// principle. Since we use truncate here, it is guaranteed that any element
 	// on the boundary will automatically fall in to the window to the right
 	// of the boundary thereby satisfying the requirement.
-	return []window.AlignedWindow{
+	return []window.AlignedKeyedWindower{
 		keyed.NewKeyedWindow(start, end),
 	}
 }
 
 // CreateWindow adds a window for a given interval window
-func (f *Fixed) CreateWindow(kw window.AlignedWindow) window.AlignedWindow {
+func (f *Fixed) CreateWindow(kw window.AlignedKeyedWindower) window.AlignedKeyedWindower {
 	f.lock.Lock()
 	defer f.lock.Unlock()
 
@@ -88,8 +88,8 @@ func (f *Fixed) CreateWindow(kw window.AlignedWindow) window.AlignedWindow {
 		return kw
 	}
 
-	earliestWindow := f.entries.Front().Value.(*keyed.KeyedWindow)
-	recentWindow := f.entries.Back().Value.(*keyed.KeyedWindow)
+	earliestWindow := f.entries.Front().Value.(*keyed.AlignedKeyedWindow)
+	recentWindow := f.entries.Back().Value.(*keyed.AlignedKeyedWindow)
 
 	// late arrival
 	if !earliestWindow.StartTime().Before(kw.EndTime()) {
@@ -100,7 +100,7 @@ func (f *Fixed) CreateWindow(kw window.AlignedWindow) window.AlignedWindow {
 	} else {
 		// a window in the middle
 		for e := f.entries.Back(); e != nil; e = e.Prev() {
-			win := e.Value.(*keyed.KeyedWindow)
+			win := e.Value.(*keyed.AlignedKeyedWindow)
 			if !win.StartTime().Before(kw.EndTime()) {
 				f.entries.InsertBefore(kw, e)
 				break
@@ -111,7 +111,7 @@ func (f *Fixed) CreateWindow(kw window.AlignedWindow) window.AlignedWindow {
 }
 
 // GetWindow returns an existing window for the given interval
-func (f *Fixed) GetWindow(kw window.AlignedWindow) window.AlignedWindow {
+func (f *Fixed) GetWindow(kw window.AlignedKeyedWindower) window.AlignedKeyedWindower {
 	f.lock.RLock()
 	defer f.lock.RUnlock()
 
@@ -121,21 +121,21 @@ func (f *Fixed) GetWindow(kw window.AlignedWindow) window.AlignedWindow {
 
 	// are we looking for a window that is later than the current latest?
 	latest := f.entries.Back()
-	lkw := latest.Value.(*keyed.KeyedWindow)
+	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.KeyedWindow)
+	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.KeyedWindow)
+		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()) {
@@ -149,14 +149,14 @@ func (f *Fixed) GetWindow(kw window.AlignedWindow) window.AlignedWindow {
 
 // RemoveWindows returns an array of keyed windows that are before the current watermark.
 // So these windows can be closed.
-func (f *Fixed) RemoveWindows(wm time.Time) []window.AlignedWindow {
+func (f *Fixed) RemoveWindows(wm time.Time) []window.AlignedKeyedWindower {
 	f.lock.Lock()
 	defer f.lock.Unlock()
 
-	closedWindows := make([]window.AlignedWindow, 0)
+	closedWindows := make([]window.AlignedKeyedWindower, 0)
 
 	for e := f.entries.Front(); e != nil; {
-		win := e.Value.(*keyed.KeyedWindow)
+		win := e.Value.(*keyed.AlignedKeyedWindow)
 		next := e.Next()
 		// remove window only after the watermark has passed the end of the window
 		if win.EndTime().Before(wm) {
diff --git a/pkg/window/windower.go b/pkg/window/windower.go
index c82067b5..b6d17959 100644
--- a/pkg/window/windower.go
+++ b/pkg/window/windower.go
@@ -22,12 +22,9 @@ import (
 	"github.com/numaproj/numaflow/pkg/pbq/partition"
 )
 
-// AlignedWindow interface represents a bounded window at a moment in time
-// for example in case of fixed and sliding windows, AlignedWindow will have the
-// same start and end time as the initial window that element is slotted in to.
-// However, same cannot be said about a session window. Window Boundaries will keep
-// changing up until a session is closed or times out.
-type AlignedWindow interface {
+// AlignedKeyedWindower represents a bounded window (i.e., it will have a definite start and end time), and the
+// keys that also fall into the same window.
+type AlignedKeyedWindower interface {
 	// StartTime returns the start time of the window
 	StartTime() time.Time
 	// EndTime returns the end time of the window
@@ -40,15 +37,15 @@ type AlignedWindow interface {
 	Keys() []string
 }
 
-// Windower manages windows
+// Windower manages AlignedKeyedWindower
 // Will be implemented by each of the windowing strategies.
 type Windower interface {
 	// AssignWindow assigns the event to the window based on give window configuration.
-	AssignWindow(eventTime time.Time) []AlignedWindow
+	AssignWindow(eventTime time.Time) []AlignedKeyedWindower
 	// CreateWindow creates a window for a supplied interval
-	CreateWindow(aw AlignedWindow) AlignedWindow
+	CreateWindow(aw AlignedKeyedWindower) AlignedKeyedWindower
 	// GetWindow returns a keyed window for a supplied interval
-	GetWindow(aw AlignedWindow) AlignedWindow
+	GetWindow(aw AlignedKeyedWindower) AlignedKeyedWindower
 	// RemoveWindows returns list of window(s) that can be closed
-	RemoveWindows(time time.Time) []AlignedWindow
+	RemoveWindows(time time.Time) []AlignedKeyedWindower
 }


Test patch:
diff --git a/pkg/watermark/ot/ot_test.go b/pkg/watermark/ot/ot_test.go
index b2d39a55..40c798b5 100644
--- a/pkg/watermark/ot/ot_test.go
+++ b/pkg/watermark/ot/ot_test.go
@@ -1,3 +1,19 @@
+/*
+Copyright 2022 The Numaproj Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
 package ot
 
 import (
diff --git a/pkg/window/keyed/keyed_test.go b/pkg/window/keyed/keyed_test.go
index e396d745..4a29a2a5 100644
--- a/pkg/window/keyed/keyed_test.go
+++ b/pkg/window/keyed/keyed_test.go
@@ -30,19 +30,19 @@ func TestKeyedWindow_AddKey(t *testing.T) {
 	kw := NewKeyedWindow(time.Unix(60, 0), time.Unix(120, 0))
 	tests := []struct {
 		name         string
-		given        *KeyedWindow
+		given        *AlignedKeyedWindow
 		input        string
 		expectedKeys map[string]struct{}
 	}{
 		{
 			name:         "no_keys",
-			given:        &KeyedWindow{},
+			given:        &AlignedKeyedWindow{},
 			input:        "key1",
 			expectedKeys: map[string]struct{}{"key1": {}},
 		},
 		{
 			name: "with_some_existing_keys",
-			given: &KeyedWindow{
+			given: &AlignedKeyedWindow{
 				keys: map[string]struct{}{"key2": {}, "key3": {}},
 			},
 			input:        "key4",
@@ -70,18 +70,18 @@ func TestKeyedWindow_Partitions(t *testing.T) {
 	kw := NewKeyedWindow(time.Unix(60, 0), time.Unix(120, 0))
 	tests := []struct {
 		name     string
-		given    *KeyedWindow
+		given    *AlignedKeyedWindow
 		input    string
 		expected []partition.ID
 	}{
 		{
 			name:     "no_keys",
-			given:    &KeyedWindow{},
+			given:    &AlignedKeyedWindow{},
 			expected: []partition.ID{},
 		},
 		{
 			name: "with_some_existing_keys",
-			given: &KeyedWindow{
+			given: &AlignedKeyedWindow{
 				keys: map[string]struct{}{"key2": {}, "key3": {}, "key4": {}},
 			},
 			expected: []partition.ID{
diff --git a/pkg/window/strategy/fixed/fixed_test.go b/pkg/window/strategy/fixed/fixed_test.go
index 6ad28425..4446c175 100644
--- a/pkg/window/strategy/fixed/fixed_test.go
+++ b/pkg/window/strategy/fixed/fixed_test.go
@@ -35,13 +35,13 @@ func TestFixed_AssignWindow(t *testing.T) {
 		name      string
 		length    time.Duration
 		eventTime time.Time
-		want      []*keyed.KeyedWindow
+		want      []*keyed.AlignedKeyedWindow
 	}{
 		{
 			name:      "minute",
 			length:    time.Minute,
 			eventTime: baseTime,
-			want: []*keyed.KeyedWindow{
+			want: []*keyed.AlignedKeyedWindow{
 				{
 					Start: time.Unix(1651129200, 0).In(loc),
 					End:   time.Unix(1651129260, 0).In(loc),
@@ -52,7 +52,7 @@ func TestFixed_AssignWindow(t *testing.T) {
 			name:      "hour",
 			length:    time.Hour,
 			eventTime: baseTime,
-			want: []*keyed.KeyedWindow{
+			want: []*keyed.AlignedKeyedWindow{
 				{
 					Start: time.Unix(1651129200, 0).In(loc),
 					End:   time.Unix(1651129200+3600, 0).In(loc),
@@ -63,7 +63,7 @@ func TestFixed_AssignWindow(t *testing.T) {
 			name:      "5_minute",
 			length:    time.Minute * 5,
 			eventTime: baseTime,
-			want: []*keyed.KeyedWindow{
+			want: []*keyed.AlignedKeyedWindow{
 				{
 					Start: time.Unix(1651129200, 0).In(loc),
 					End:   time.Unix(1651129200+300, 0).In(loc),
@@ -74,7 +74,7 @@ func TestFixed_AssignWindow(t *testing.T) {
 			name:      "30_second",
 			length:    time.Second * 30,
 			eventTime: baseTime,
-			want: []*keyed.KeyedWindow{
+			want: []*keyed.AlignedKeyedWindow{
 				{
 					Start: time.Unix(1651129200, 0).In(loc),
 					End:   time.Unix(1651129230, 0).In(loc),
@@ -98,18 +98,18 @@ func TestAligned_CreateWindow(t *testing.T) {
 	windows := NewFixed(60 * time.Second)
 	tests := []struct {
 		name            string
-		given           []*keyed.KeyedWindow
-		input           *keyed.KeyedWindow
-		expectedWindows []*keyed.KeyedWindow
+		given           []*keyed.AlignedKeyedWindow
+		input           *keyed.AlignedKeyedWindow
+		expectedWindows []*keyed.AlignedKeyedWindow
 	}{
 		{
 			name:  "FirstWindow",
-			given: []*keyed.KeyedWindow{},
-			input: &keyed.KeyedWindow{
+			given: []*keyed.AlignedKeyedWindow{},
+			input: &keyed.AlignedKeyedWindow{
 				Start: time.Unix(0, 0),
 				End:   time.Unix(60, 0),
 			},
-			expectedWindows: []*keyed.KeyedWindow{
+			expectedWindows: []*keyed.AlignedKeyedWindow{
 				{
 					Start: time.Unix(0, 0),
 					End:   time.Unix(60, 0),
@@ -118,17 +118,17 @@ func TestAligned_CreateWindow(t *testing.T) {
 		},
 		{
 			name: "late_window",
-			given: []*keyed.KeyedWindow{
+			given: []*keyed.AlignedKeyedWindow{
 				{
 					Start: time.Unix(120, 0),
 					End:   time.Unix(180, 0),
 				},
 			},
-			input: &keyed.KeyedWindow{
+			input: &keyed.AlignedKeyedWindow{
 				Start: time.Unix(60, 0),
 				End:   time.Unix(120, 0),
 			},
-			expectedWindows: []*keyed.KeyedWindow{
+			expectedWindows: []*keyed.AlignedKeyedWindow{
 				{
 					Start: time.Unix(60, 0),
 					End:   time.Unix(120, 0),
@@ -141,17 +141,17 @@ func TestAligned_CreateWindow(t *testing.T) {
 		},
 		{
 			name: "early_window",
-			given: []*keyed.KeyedWindow{
+			given: []*keyed.AlignedKeyedWindow{
 				{
 					Start: time.Unix(120, 0),
 					End:   time.Unix(180, 0),
 				},
 			},
-			input: &keyed.KeyedWindow{
+			input: &keyed.AlignedKeyedWindow{
 				Start: time.Unix(240, 0),
 				End:   time.Unix(300, 0),
 			},
-			expectedWindows: []*keyed.KeyedWindow{
+			expectedWindows: []*keyed.AlignedKeyedWindow{
 				{
 					Start: time.Unix(120, 0),
 					End:   time.Unix(180, 0),
@@ -164,7 +164,7 @@ func TestAligned_CreateWindow(t *testing.T) {
 		},
 		{
 			name: "insert_middle",
-			given: []*keyed.KeyedWindow{
+			given: []*keyed.AlignedKeyedWindow{
 				{
 					Start: time.Unix(120, 0),
 					End:   time.Unix(180, 0),
@@ -174,11 +174,11 @@ func TestAligned_CreateWindow(t *testing.T) {
 					End:   time.Unix(300, 0),
 				},
 			},
-			input: &keyed.KeyedWindow{
+			input: &keyed.AlignedKeyedWindow{
 				Start: time.Unix(180, 0),
 				End:   time.Unix(240, 0),
 			},
-			expectedWindows: []*keyed.KeyedWindow{
+			expectedWindows: []*keyed.AlignedKeyedWindow{
 				{
 					Start: time.Unix(120, 0),
 					End:   time.Unix(180, 0),
@@ -205,8 +205,8 @@ func TestAligned_CreateWindow(t *testing.T) {
 			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.KeyedWindow).StartTime())
-				assert.Equal(t, kw.EndTime(), node.Value.(*keyed.KeyedWindow).EndTime())
+				assert.Equal(t, kw.StartTime(), node.Value.(*keyed.AlignedKeyedWindow).StartTime())
+				assert.Equal(t, kw.EndTime(), node.Value.(*keyed.AlignedKeyedWindow).EndTime())
 				node = node.Next()
 			}
 		})
@@ -217,14 +217,14 @@ func TestAligned_GetWindow(t *testing.T) {
 	windows := NewFixed(60 * time.Second)
 	tests := []struct {
 		name           string
-		given          []*keyed.KeyedWindow
-		input          *keyed.KeyedWindow
-		expectedWindow *keyed.KeyedWindow
+		given          []*keyed.AlignedKeyedWindow
+		input          *keyed.AlignedKeyedWindow
+		expectedWindow *keyed.AlignedKeyedWindow
 	}{
 		{
 			name:  "non_existing",
-			given: []*keyed.KeyedWindow{},
-			input: &keyed.KeyedWindow{
+			given: []*keyed.AlignedKeyedWindow{},
+			input: &keyed.AlignedKeyedWindow{
 				Start: time.Unix(0, 0),
 				End:   time.Unix(60, 0),
 			},
@@ -232,7 +232,7 @@ func TestAligned_GetWindow(t *testing.T) {
 		},
 		{
 			name: "non_existing_before_earlier",
-			given: []*keyed.KeyedWindow{
+			given: []*keyed.AlignedKeyedWindow{
 				{
 					Start: time.Unix(60, 0),
 					End:   time.Unix(120, 0),
@@ -242,7 +242,7 @@ func TestAligned_GetWindow(t *testing.T) {
 					End:   time.Unix(180, 0),
 				},
 			},
-			input: &keyed.KeyedWindow{
+			input: &keyed.AlignedKeyedWindow{
 				Start: time.Unix(0, 0),
 				End:   time.Unix(60, 0),
 			},
@@ -250,7 +250,7 @@ func TestAligned_GetWindow(t *testing.T) {
 		},
 		{
 			name: "non_existing_after_recent",
-			given: []*keyed.KeyedWindow{
+			given: []*keyed.AlignedKeyedWindow{
 				{
 					Start: time.Unix(60, 0),
 					End:   time.Unix(120, 0),
@@ -260,7 +260,7 @@ func TestAligned_GetWindow(t *testing.T) {
 					End:   time.Unix(180, 0),
 				},
 			},
-			input: &keyed.KeyedWindow{
+			input: &keyed.AlignedKeyedWindow{
 				Start: time.Unix(180, 0),
 				End:   time.Unix(240, 0),
 			},
@@ -268,7 +268,7 @@ func TestAligned_GetWindow(t *testing.T) {
 		},
 		{
 			name: "non_existing_middle",
-			given: []*keyed.KeyedWindow{
+			given: []*keyed.AlignedKeyedWindow{
 				{
 					Start: time.Unix(60, 0),
 					End:   time.Unix(120, 0),
@@ -278,7 +278,7 @@ func TestAligned_GetWindow(t *testing.T) {
 					End:   time.Unix(240, 0),
 				},
 			},
-			input: &keyed.KeyedWindow{
+			input: &keyed.AlignedKeyedWindow{
 				Start: time.Unix(120, 0),
 				End:   time.Unix(180, 0),
 			},
@@ -286,7 +286,7 @@ func TestAligned_GetWindow(t *testing.T) {
 		},
 		{
 			name: "existing",
-			given: []*keyed.KeyedWindow{
+			given: []*keyed.AlignedKeyedWindow{
 				{
 					Start: time.Unix(60, 0),
 					End:   time.Unix(120, 0),
@@ -296,18 +296,18 @@ func TestAligned_GetWindow(t *testing.T) {
 					End:   time.Unix(180, 0),
 				},
 			},
-			input: &keyed.KeyedWindow{
+			input: &keyed.AlignedKeyedWindow{
 				Start: time.Unix(60, 0),
 				End:   time.Unix(120, 0),
 			},
-			expectedWindow: &keyed.KeyedWindow{
+			expectedWindow: &keyed.AlignedKeyedWindow{
 				Start: time.Unix(60, 0),
 				End:   time.Unix(120, 0),
 			},
 		},
 		{
 			name: "first_window",
-			given: []*keyed.KeyedWindow{
+			given: []*keyed.AlignedKeyedWindow{
 				{
 					Start: time.Unix(120, 0),
 					End:   time.Unix(180, 0),
@@ -317,18 +317,18 @@ func TestAligned_GetWindow(t *testing.T) {
 					End:   time.Unix(300, 0),
 				},
 			},
-			input: &keyed.KeyedWindow{
+			input: &keyed.AlignedKeyedWindow{
 				Start: time.Unix(120, 0),
 				End:   time.Unix(180, 0),
 			},
-			expectedWindow: &keyed.KeyedWindow{
+			expectedWindow: &keyed.AlignedKeyedWindow{
 				Start: time.Unix(120, 0),
 				End:   time.Unix(180, 0),
 			},
 		},
 		{
 			name: "last_window",
-			given: []*keyed.KeyedWindow{
+			given: []*keyed.AlignedKeyedWindow{
 				{
 					Start: time.Unix(120, 0),
 					End:   time.Unix(180, 0),
@@ -338,11 +338,11 @@ func TestAligned_GetWindow(t *testing.T) {
 					End:   time.Unix(300, 0),
 				},
 			},
-			input: &keyed.KeyedWindow{
+			input: &keyed.AlignedKeyedWindow{
 				Start: time.Unix(240, 0),
 				End:   time.Unix(300, 0),
 			},
-			expectedWindow: &keyed.KeyedWindow{
+			expectedWindow: &keyed.AlignedKeyedWindow{
 				Start: time.Unix(240, 0),
 				End:   time.Unix(300, 0),
 			},
@@ -369,37 +369,37 @@ func TestAligned_RemoveWindow(t *testing.T) {
 	windows := NewFixed(60 * time.Second)
 	tests := []struct {
 		name            string
-		given           []*keyed.KeyedWindow
+		given           []*keyed.AlignedKeyedWindow
 		input           time.Time
-		expectedWindows []*keyed.KeyedWindow
+		expectedWindows []*keyed.AlignedKeyedWindow
 	}{
 		{
 			name:            "empty_windows",
-			given:           []*keyed.KeyedWindow{},
+			given:           []*keyed.AlignedKeyedWindow{},
 			input:           time.Unix(60, 0),
-			expectedWindows: []*keyed.KeyedWindow{},
+			expectedWindows: []*keyed.AlignedKeyedWindow{},
 		},
 		{
 			name: "wm_on_edge",
-			given: []*keyed.KeyedWindow{
+			given: []*keyed.AlignedKeyedWindow{
 				{
 					Start: time.Unix(120, 0),
 					End:   time.Unix(180, 0),
 				},
 			},
 			input:           time.Unix(180, 0),
-			expectedWindows: []*keyed.KeyedWindow{},
+			expectedWindows: []*keyed.AlignedKeyedWindow{},
 		},
 		{
 			name: "single_window",
-			given: []*keyed.KeyedWindow{
+			given: []*keyed.AlignedKeyedWindow{
 				{
 					Start: time.Unix(120, 0),
 					End:   time.Unix(180, 0),
 				},
 			},
 			input: time.Unix(181, 0),
-			expectedWindows: []*keyed.KeyedWindow{
+			expectedWindows: []*keyed.AlignedKeyedWindow{
 				{
 					Start: time.Unix(120, 0),
 					End:   time.Unix(180, 0),
@@ -408,7 +408,7 @@ func TestAligned_RemoveWindow(t *testing.T) {
 		},
 		{
 			name: "single_when_multiple",
-			given: []*keyed.KeyedWindow{
+			given: []*keyed.AlignedKeyedWindow{
 				{
 					Start: time.Unix(120, 0),
 					End:   time.Unix(180, 0),
@@ -423,7 +423,7 @@ func TestAligned_RemoveWindow(t *testing.T) {
 				},
 			},
 			input: time.Unix(181, 0),
-			expectedWindows: []*keyed.KeyedWindow{
+			expectedWindows: []*keyed.AlignedKeyedWindow{
 				{
 					Start: time.Unix(120, 0),
 					End:   time.Unix(180, 0),
@@ -432,7 +432,7 @@ func TestAligned_RemoveWindow(t *testing.T) {
 		},
 		{
 			name: "multiple_removals",
-			given: []*keyed.KeyedWindow{
+			given: []*keyed.AlignedKeyedWindow{
 				{
 					Start: time.Unix(120, 0),
 					End:   time.Unix(180, 0),
@@ -447,7 +447,7 @@ func TestAligned_RemoveWindow(t *testing.T) {
 				},
 			},
 			input: time.Unix(245, 0),
-			expectedWindows: []*keyed.KeyedWindow{
+			expectedWindows: []*keyed.AlignedKeyedWindow{
 				{
 					Start: time.Unix(120, 0),
 					End:   time.Unix(180, 0),
@@ -473,7 +473,7 @@ func TestAligned_RemoveWindow(t *testing.T) {
 	}
 }
 
-func setup(windows *Fixed, wins []*keyed.KeyedWindow) {
+func setup(windows *Fixed, wins []*keyed.AlignedKeyedWindow) {
 	windows.entries = list.New()
 	for _, win := range wins {
 		windows.entries.PushBack(win)