Repository: numaproj/numaflow
Pull request #1475 (2024-01-22 14:00:14)

chore: minor TimedWindow refactor


Patch:
diff --git a/pkg/apis/numaflow/v1alpha1/const.go b/pkg/apis/numaflow/v1alpha1/const.go
index 2da1ef0d..b3178a63 100644
--- a/pkg/apis/numaflow/v1alpha1/const.go
+++ b/pkg/apis/numaflow/v1alpha1/const.go
@@ -164,6 +164,9 @@ const (
 	// DefaultKeyForNonKeyedData Default key for non keyed stream
 	DefaultKeyForNonKeyedData = "NON_KEYED_STREAM"
 
+	// KeysDelimitter is the delimitter used to join keys
+	KeysDelimitter = ":"
+
 	// UDF map streaming
 	MapUdfStreamKey = "numaflow.numaproj.io/map-stream"
 
diff --git a/pkg/reduce/data_forward.go b/pkg/reduce/data_forward.go
index 90b162f1..c3fa8ca7 100644
--- a/pkg/reduce/data_forward.go
+++ b/pkg/reduce/data_forward.go
@@ -163,7 +163,7 @@ func (df *DataForward) ReplayPersistedMessages(ctx context.Context) error {
 		// create a window for each partition and insert it to the windower
 		// so that the window can be closed when the watermark
 		// crosses the window.
-		var timedWindow = window.NewWindowFromPartition(&p)
+		var timedWindow = window.NewAlignedTimedWindow(p.Start, p.End, p.Slot)
 
 		df.windower.InsertWindow(timedWindow)
 
diff --git a/pkg/reduce/pbq/pbq.go b/pkg/reduce/pbq/pbq.go
index 23b48657..31d631c7 100644
--- a/pkg/reduce/pbq/pbq.go
+++ b/pkg/reduce/pbq/pbq.go
@@ -145,7 +145,7 @@ readLoop:
 			// FIXME: support for session window
 			var w window.TimedWindow
 			if p.windowType == window.Aligned {
-				w = window.NewWindowFromPartition(&p.PartitionID)
+				w = window.NewAlignedTimedWindow(p.PartitionID.Start, p.PartitionID.End, p.PartitionID.Slot)
 			} else {
 				p.log.Errorw("unAligned window strategy not supported", zap.Any("ID", p.PartitionID), zap.Any("strategy", p.windowType))
 			}
diff --git a/pkg/reduce/pnf/processandforward.go b/pkg/reduce/pnf/processandforward.go
index a20d346f..42ad8220 100644
--- a/pkg/reduce/pnf/processandforward.go
+++ b/pkg/reduce/pnf/processandforward.go
@@ -151,7 +151,7 @@ outerLoop:
 				// FIXME(session): we need to compact the pbq for unAligned when we have received the EOF response from the UDF.
 				// we should not use p.partitionId here, we should use the partition id from the response.
 				// because for unaligned p.partitionId indicates the shared partition id for the key.
-				p.publishWM(ctx, response.Window.Partition())
+				p.publishWM(ctx, response.Window.EndTime())
 
 				// delete the closed windows which are tracked by the windower
 				p.windower.DeleteClosedWindow(response.Window)
@@ -190,9 +190,9 @@ outerLoop:
 
 	// for aligned we don't track the windows for every key, we need to delete the window
 	// once we have received all the responses from the UDF.
-	p.publishWM(ctx, p.partitionId)
+	p.publishWM(ctx, p.partitionId.End)
 	// delete the closed windows which are tracked by the windower
-	p.windower.DeleteClosedWindow(window.NewWindowFromPartition(p.partitionId))
+	p.windower.DeleteClosedWindow(window.NewAlignedTimedWindow(p.partitionId.Start, p.partitionId.End, p.partitionId.Slot))
 
 	// Since we have successfully processed all the messages for a window, we can now delete the persisted messages.
 	err := p.pbqReader.GC()
@@ -354,7 +354,7 @@ func (p *processAndForward) writeToBuffer(ctx context.Context, edgeName string,
 
 // publishWM publishes the watermark to each edge.
 // TODO: support multi partitioned edges.
-func (p *processAndForward) publishWM(ctx context.Context, id *partition.ID) {
+func (p *processAndForward) publishWM(ctx context.Context, endTime time.Time) {
 	// publish watermark, we publish window end time minus one millisecond  as watermark
 	// but if there's a window that's about to be closed which has a end time before the current window end time,
 	// we publish that window's end time as watermark. This is to ensure that the watermark is monotonically increasing.
@@ -363,7 +363,7 @@ func (p *processAndForward) publishWM(ctx context.Context, id *partition.ID) {
 	if oldestClosedWindowEndTime != time.UnixMilli(-1) && oldestClosedWindowEndTime.Before(p.partitionId.End) {
 		wm = wmb.Watermark(oldestClosedWindowEndTime.Add(-1 * time.Millisecond))
 	} else {
-		wm = wmb.Watermark(id.End.Add(-1 * time.Millisecond))
+		wm = wmb.Watermark(endTime.Add(-1 * time.Millisecond))
 	}
 
 	// activeWatermarkBuffers records the buffers that the publisher has published
diff --git a/pkg/udf/rpc/grpc_aligned_reduce.go b/pkg/udf/rpc/grpc_aligned_reduce.go
index 7a3bd10d..bb363c4a 100644
--- a/pkg/udf/rpc/grpc_aligned_reduce.go
+++ b/pkg/udf/rpc/grpc_aligned_reduce.go
@@ -251,11 +251,11 @@ func parseReduceResponse(response *reducepb.ReduceResponse) *window.TimedWindowR
 	// because for fixed and sliding we don't track keys.
 	return &window.TimedWindowResponse{
 		WriteMessage: taggedMessage,
-		Window: window.NewWindowFromPartition(&partition.ID{
-			Start: response.GetWindow().GetStart().AsTime(),
-			End:   response.GetWindow().GetEnd().AsTime(),
-			Slot:  response.GetWindow().GetSlot(),
-		}),
+		Window: window.NewAlignedTimedWindow(
+			response.GetWindow().GetStart().AsTime(),
+			response.GetWindow().GetEnd().AsTime(),
+			response.GetWindow().GetSlot(),
+		),
 		// this will be set once the SDK has sent the last message
 		EOF: response.GetEOF(),
 	}
diff --git a/pkg/udf/rpc/grpc_unaligned_reduce.go b/pkg/udf/rpc/grpc_unaligned_reduce.go
index 9bec9cf6..6fce6d9b 100644
--- a/pkg/udf/rpc/grpc_unaligned_reduce.go
+++ b/pkg/udf/rpc/grpc_unaligned_reduce.go
@@ -217,12 +217,10 @@ func parseSessionReduceResponse(response *sessionreducepb.SessionReduceResponse)
 
 	return &window.TimedWindowResponse{
 		WriteMessage: taggedMessage,
-		Window: window.NewWindowFromPartitionAndKeys(
-			&partition.ID{
-				Start: time.UnixMilli(response.GetKeyedWindow().GetStart().AsTime().UnixMilli()),
-				End:   time.UnixMilli(response.GetKeyedWindow().GetEnd().AsTime().UnixMilli()),
-				Slot:  response.GetKeyedWindow().GetSlot(),
-			},
+		Window: window.NewUnalignedTimedWindow(
+			response.GetKeyedWindow().GetStart().AsTime(),
+			response.GetKeyedWindow().GetEnd().AsTime(),
+			response.GetKeyedWindow().GetSlot(),
 			response.GetKeyedWindow().GetKeys(),
 		),
 		EOF: response.GetEOF(),
diff --git a/pkg/window/sorted_window_endtime.go b/pkg/window/sorted_window_endtime.go
index 2f9e5358..91591028 100644
--- a/pkg/window/sorted_window_endtime.go
+++ b/pkg/window/sorted_window_endtime.go
@@ -103,8 +103,7 @@ func (s *SortedWindowListByEndTime) InsertIfNotPresent(window TimedWindow) (Time
 
 	// Check if the window is already present in the list
 	for i := index; i < len(s.windows); i++ {
-		if s.windows[i].Partition().String() == window.Partition().String() &&
-			compareKeys(s.windows[i].Keys(), window.Keys()) {
+		if s.windows[i].ID() == window.ID() {
 			return s.windows[i], true
 		}
 
@@ -136,8 +135,7 @@ func (s *SortedWindowListByEndTime) Delete(window TimedWindow) (deleted bool) {
 
 	// Delete the window if it is present in the list
 	for i := index; i < len(s.windows); i++ {
-		if s.windows[i].Partition().String() == window.Partition().String() &&
-			compareKeys(s.windows[i].Keys(), window.Keys()) {
+		if s.windows[i].ID() == window.ID() {
 			s.windows = append(s.windows[:i], s.windows[i+1:]...)
 			deleted = true
 			break
diff --git a/pkg/window/strategy/fixed/fixed.go b/pkg/window/strategy/fixed/fixed.go
index d54b0bb7..6830d205 100644
--- a/pkg/window/strategy/fixed/fixed.go
+++ b/pkg/window/strategy/fixed/fixed.go
@@ -23,6 +23,7 @@ limitations under the License.
 package fixed
 
 import (
+	"fmt"
 	"strconv"
 	"time"
 
@@ -37,7 +38,9 @@ import (
 type fixedWindow struct {
 	startTime time.Time
 	endTime   time.Time
+	partition *partition.ID
 	slot      string
+	id        string
 }
 
 // NewFixedWindow returns a new window for the given message.
@@ -52,6 +55,12 @@ func NewFixedWindow(length time.Duration, message *isb.ReadMessage) window.Timed
 		startTime: start,
 		endTime:   end,
 		slot:      slot,
+		partition: &partition.ID{
+			Start: start,
+			End:   end,
+			Slot:  slot,
+		},
+		id: fmt.Sprintf("%d-%d-%s", start.UnixMilli(), end.UnixMilli(), slot),
 	}
 }
 
@@ -74,11 +83,11 @@ func (w *fixedWindow) Keys() []string {
 }
 
 func (w *fixedWindow) Partition() *partition.ID {
-	return &partition.ID{
-		Start: w.startTime,
-		End:   w.endTime,
-		Slot:  w.slot,
-	}
+	return w.partition
+}
+
+func (w *fixedWindow) ID() string {
+	return w.id
 }
 
 func (w *fixedWindow) Merge(_ window.TimedWindow) {
diff --git a/pkg/window/strategy/session/session.go b/pkg/window/strategy/session/session.go
index adb5020c..6009fee6 100644
--- a/pkg/window/strategy/session/session.go
+++ b/pkg/window/strategy/session/session.go
@@ -17,7 +17,7 @@ limitations under the License.
 package session
 
 import (
-	"math"
+	"fmt"
 	"strconv"
 	"strings"
 	"time"
@@ -29,22 +29,14 @@ import (
 	"github.com/numaproj/numaflow/pkg/window"
 )
 
-const delimiter = ":"
-
-// SharedSessionPartition is a common partition for session window.
-// session windows share a common pbq, this partition is used to identify the pbq instance.
-var SharedSessionPartition = partition.ID{
-	Start: time.UnixMilli(0),
-	End:   time.UnixMilli(math.MaxInt64),
-	Slot:  "slot-0",
-}
-
 // sessionWindow TimedWindow implementation for Session window.
 type sessionWindow struct {
 	startTime time.Time
 	endTime   time.Time
 	slot      string
 	keys      []string
+	partition *partition.ID
+	id        string
 }
 
 func NewSessionWindow(startTime time.Time, gap time.Duration, message *isb.ReadMessage) window.TimedWindow {
@@ -59,6 +51,8 @@ func NewSessionWindow(startTime time.Time, gap time.Duration, message *isb.ReadM
 		endTime:   end,
 		slot:      slot,
 		keys:      message.Keys,
+		partition: &window.SharedUnalignedPartition,
+		id:        fmt.Sprintf("%d-%d-%s-%s", start.UnixMilli(), end.UnixMilli(), slot, strings.Join(message.Keys, dfv1.KeysDelimitter)),
 	}
 }
 
@@ -81,15 +75,21 @@ func (w *sessionWindow) Keys() []string {
 }
 
 func (w *sessionWindow) Partition() *partition.ID {
-	return &partition.ID{
-		Start: w.startTime,
-		End:   w.endTime,
-		Slot:  w.slot,
-	}
+	// session windows share a common pbq, this partition is used to identify the common pbq instance.
+	return w.partition
+}
+
+func (w *sessionWindow) ID() string {
+	return w.id
 }
 
 // Merge window merges two windows. This operation of merge happens in Unaligned windows.
 func (w *sessionWindow) Merge(tw window.TimedWindow) {
+
+	if !tw.StartTime().Before(w.startTime) && !tw.EndTime().After(w.endTime) {
+		return
+	}
+
 	// expand the start and end to accommodate the new window
 	if tw.StartTime().Before(w.startTime) {
 		w.startTime = tw.StartTime()
@@ -98,6 +98,21 @@ func (w *sessionWindow) Merge(tw window.TimedWindow) {
 	if tw.EndTime().After(w.endTime) {
 		w.endTime = tw.EndTime()
 	}
+
+	// udpate the id with the new start and end time
+	w.id = fmt.Sprintf("%d-%d-%s-%s", w.startTime.UnixMilli(), w.endTime.UnixMilli(), w.slot, strings.Join(w.keys, dfv1.KeysDelimitter))
+}
+
+// Expand expands the window end time to the new endTime. An interesting property of Unaligned windows :).
+func (w *sessionWindow) Expand(endTime time.Time) {
+	if !endTime.After(w.endTime) {
+		return
+	}
+
+	w.endTime = endTime
+	// udpate the id with the new end time
+	w.id = fmt.Sprintf("%d-%d-%s-%s", w.startTime.UnixMilli(), w.endTime.UnixMilli(), w.slot, strings.Join(w.keys, dfv1.KeysDelimitter))
+
 }
 
 // cloneWindow makes a deep copy of the window.
@@ -110,13 +125,6 @@ func cloneWindow(win window.TimedWindow) *sessionWindow {
 	}
 }
 
-// Expand expands the window end time to the new endTime. An interesting property of Unaligned windows :).
-func (w *sessionWindow) Expand(endTime time.Time) {
-	if endTime.After(w.endTime) {
-		w.endTime = endTime
-	}
-}
-
 // Windower is an implementation of TimedWindower of session window, windower is responsible for assigning
 // windows to the incoming messages and closing the windows that are past the watermark.
 type Windower struct {
@@ -167,7 +175,7 @@ func (*Windower) Type() window.Type {
 // - Append to an existing window (the message has the event-time such that gap + event-time is < window end time).
 func (w *Windower) AssignWindows(message *isb.ReadMessage) []*window.TimedWindowRequest {
 	var (
-		combinedKey      = strings.Join(message.Keys, delimiter)
+		combinedKey      = strings.Join(message.Keys, dfv1.KeysDelimitter)
 		windowOperations = make([]*window.TimedWindowRequest, 0)
 		win              = NewSessionWindow(message.EventTime, w.gap, message)
 	)
@@ -182,7 +190,7 @@ func (w *Windower) AssignWindows(message *isb.ReadMessage) []*window.TimedWindow
 		list = window.NewSortedWindowListByEndTime()
 		// since it's the first window, we can insert it at the front
 		list.InsertFront(win)
-		windowOperations = append(windowOperations, createWindowOperation(message, window.Open, []window.TimedWindow{win}, &SharedSessionPartition))
+		windowOperations = append(windowOperations, createWindowOperation(message, window.Open, []window.TimedWindow{win}, &window.SharedUnalignedPartition))
 		w.activeWindows[combinedKey] = list
 
 		return windowOperations
@@ -208,11 +216,11 @@ func (w *Windower) AssignWindows(message *isb.ReadMessage) []*window.TimedWindow
 
 			// update the reference of the window in the list with the expanded window
 			windowToBeMerged.Merge(win)
-			windowOperations = append(windowOperations, createWindowOperation(message, window.Expand, []window.TimedWindow{oldWindow, windowToBeMerged}, &SharedSessionPartition))
+			windowOperations = append(windowOperations, createWindowOperation(message, window.Expand, []window.TimedWindow{oldWindow, windowToBeMerged}, &window.SharedUnalignedPartition))
 		} else { // no need to expand, message doesn't change the window end-time.
 			// if the returned window is same as the window created by the message, that means
 			// the message belongs to an existing window and it's an append operation
-			windowOperations = append(windowOperations, createWindowOperation(message, window.Append, []window.TimedWindow{windowToBeMerged}, &SharedSessionPartition))
+			windowOperations = append(windowOperations, createWindowOperation(message, window.Append, []window.TimedWindow{windowToBeMerged}, &window.SharedUnalignedPartition))
 		}
 
 		return windowOperations
@@ -220,14 +228,14 @@ func (w *Windower) AssignWindows(message *isb.ReadMessage) []*window.TimedWindow
 
 	// if the window cannot be merged, that means we need to create a new window to the active windows list
 	list.Insert(win)
-	windowOperations = append(windowOperations, createWindowOperation(message, window.Open, []window.TimedWindow{win}, &SharedSessionPartition))
+	windowOperations = append(windowOperations, createWindowOperation(message, window.Open, []window.TimedWindow{win}, &window.SharedUnalignedPartition))
 
 	return windowOperations
 }
 
 // InsertWindow inserts a window to the list of active windows.
 func (w *Windower) InsertWindow(tw window.TimedWindow) {
-	combinedKey := strings.Join(tw.Keys(), delimiter)
+	combinedKey := strings.Join(tw.Keys(), dfv1.KeysDelimitter)
 	if list, ok := w.activeWindows[combinedKey]; !ok {
 		list = window.NewSortedWindowListByEndTime()
 		list.InsertFront(tw)
@@ -286,7 +294,7 @@ func (w *Windower) CloseWindows(time time.Time) []*window.TimedWindowRequest {
 			if len(windows) > 1 {
 
 				// make UDF aware that windows have to be merged
-				windowOperations = append(windowOperations, createWindowOperation(nil, window.Merge, windows, &SharedSessionPartition))
+				windowOperations = append(windowOperations, createWindowOperation(nil, window.Merge, windows, &window.SharedUnalignedPartition))
 
 				// merge the first window with subsequent windows and it grows as it merges
 				var mergedWindow = cloneWindow(windows[0])
@@ -334,19 +342,19 @@ func (w *Windower) handleWindowToBeMerged(canBeMerged bool, toBeMerged window.Ti
 	if canBeMerged {
 		oldWindow := cloneWindow(toBeMerged)
 		toBeMerged.Merge(latestWin)
-		return createWindowOperation(nil, window.Merge, []window.TimedWindow{latestWin, oldWindow}, &SharedSessionPartition)
+		return createWindowOperation(nil, window.Merge, []window.TimedWindow{latestWin, oldWindow}, &window.SharedUnalignedPartition)
 	}
 
 	w.closedWindows.Insert(latestWin)
-	return createWindowOperation(nil, window.Close, []window.TimedWindow{latestWin}, &SharedSessionPartition)
+	return createWindowOperation(nil, window.Close, []window.TimedWindow{latestWin}, &window.SharedUnalignedPartition)
 }
 
 // NextWindowToBeClosed returns the next window yet to be closed.
 func (w *Windower) NextWindowToBeClosed() window.TimedWindow {
 	return &sessionWindow{
-		startTime: SharedSessionPartition.Start,
-		endTime:   SharedSessionPartition.End,
-		slot:      SharedSessionPartition.Slot,
+		startTime: window.SharedUnalignedPartition.Start,
+		endTime:   window.SharedUnalignedPartition.End,
+		slot:      window.SharedUnalignedPartition.Slot,
 	}
 }
 
diff --git a/pkg/window/strategy/sliding/sliding.go b/pkg/window/strategy/sliding/sliding.go
index 5a119b61..ad2703e3 100644
--- a/pkg/window/strategy/sliding/sliding.go
+++ b/pkg/window/strategy/sliding/sliding.go
@@ -22,6 +22,7 @@ limitations under the License.
 package sliding
 
 import (
+	"fmt"
 	"strconv"
 	"time"
 
@@ -37,6 +38,8 @@ type slidingWindow struct {
 	startTime time.Time
 	endTime   time.Time
 	slot      string
+	partition *partition.ID
+	id        string
 }
 
 func NewSlidingWindow(startTime time.Time, endTime time.Time) window.TimedWindow {
@@ -50,6 +53,12 @@ func NewSlidingWindow(startTime time.Time, endTime time.Time) window.TimedWindow
 		startTime: startTime,
 		endTime:   endTime,
 		slot:      slot,
+		partition: &partition.ID{
+			Start: startTime,
+			End:   endTime,
+			Slot:  slot,
+		},
+		id: fmt.Sprintf("%d-%d-%s", startTime.UnixMilli(), endTime.UnixMilli(), slot),
 	}
 }
 
@@ -72,11 +81,11 @@ func (w *slidingWindow) Keys() []string {
 }
 
 func (w *slidingWindow) Partition() *partition.ID {
-	return &partition.ID{
-		Start: w.startTime,
-		End:   w.endTime,
-		Slot:  w.slot,
-	}
+	return w.partition
+}
+
+func (w *slidingWindow) ID() string {
+	return w.id
 }
 
 // Merge merges the given window with the current window.
diff --git a/pkg/window/windower.go b/pkg/window/windower.go
index efb5079f..f96b2c94 100644
--- a/pkg/window/windower.go
+++ b/pkg/window/windower.go
@@ -17,6 +17,9 @@ limitations under the License.
 package window
 
 import (
+	"fmt"
+	"math"
+	"strings"
 	"time"
 
 	"github.com/numaproj/numaflow/pkg/isb"
@@ -57,12 +60,17 @@ type TimedWindow interface {
 	Slot() string
 	// Partition returns the partition id of the window, partition is
 	// combination of start time, end time and slot.
-	// for aligned windows, this will be used to map to the pbq instance
-	// for unaligned windows, we use a single pbq instance for all the windows
+	// This will be used to map to the pbq instance where the messages
+	// should be persisted.
 	Partition() *partition.ID
 	// Keys returns the keys of the window tracked for Unaligned windows.
 	// This will return empty for Aligned windows.
 	Keys() []string
+	// ID returns the id which is the unique identifier for the window.
+	// This is used to compare the windows. For Aligned windows, this is the
+	// combination of start time, end time and slot. For Unaligned windows,
+	// this is the combination of start time, end time, slot and keys.
+	ID() string
 	// Merge merges the window with the new window. It is used only for
 	// Unaligned window.
 	Merge(tw TimedWindow)
@@ -71,58 +79,124 @@ type TimedWindow interface {
 	Expand(endTime time.Time)
 }
 
-// timedWindow implements TimedWindow.
-type timedWindow struct {
+// SharedUnalignedPartition is a common partition for unaligned window.
+// unaligned windows share a common pbq, this partition is used to identify the pbq instance.
+var SharedUnalignedPartition = partition.ID{
+	Start: time.UnixMilli(0),
+	End:   time.UnixMilli(math.MaxInt64),
+	Slot:  "slot-0",
+}
+
+type alignedTimedWindow struct {
 	startTime time.Time
 	endTime   time.Time
 	slot      string
-	keys      []string
+	partition *partition.ID
+	id        string
 }
 
-// NewWindowFromPartition returns a new TimedWindow for the given partition id.
-func NewWindowFromPartition(id *partition.ID) TimedWindow {
-	return &timedWindow{
-		startTime: id.Start,
-		endTime:   id.End,
-		slot:      id.Slot,
+// NewAlignedTimedWindow returns a new TimedWindow for the given start time, end time and slot.
+func NewAlignedTimedWindow(st time.Time, et time.Time, slot string) TimedWindow {
+	return &alignedTimedWindow{
+		startTime: st,
+		endTime:   et,
+		slot:      slot,
+		partition: &partition.ID{
+			Start: st,
+			End:   et,
+			Slot:  slot,
+		},
+		id: fmt.Sprintf("%d-%d-%s", st.UnixMilli(), et.UnixMilli(), slot),
 	}
 }
 
-// NewWindowFromPartitionAndKeys returns a new TimedWindow for the given partition id and keys.
-func NewWindowFromPartitionAndKeys(id *partition.ID, keys []string) TimedWindow {
-	return &timedWindow{
-		startTime: id.Start,
-		endTime:   id.End,
-		slot:      id.Slot,
+func (at *alignedTimedWindow) StartTime() time.Time {
+	return at.startTime
+}
+
+func (at *alignedTimedWindow) EndTime() time.Time {
+	return at.endTime
+}
+
+func (at *alignedTimedWindow) Slot() string {
+	return at.slot
+}
+
+func (at *alignedTimedWindow) Partition() *partition.ID {
+	return at.partition
+}
+
+func (at *alignedTimedWindow) Keys() []string {
+	return nil
+}
+
+func (at *alignedTimedWindow) ID() string {
+	return at.id
+}
+
+func (at *alignedTimedWindow) Merge(tw TimedWindow) {
+	// never invoked for Aligned Window
+}
+
+func (at *alignedTimedWindow) Expand(endTime time.Time) {
+	// never invoked for Aligned Window
+}
+
+// timedWindow implements TimedWindow.
+type unalignedTimedWindow struct {
+	startTime time.Time
+	endTime   time.Time
+	slot      string
+	keys      []string
+	partition *partition.ID
+	id        string
+}
+
+// NewUnalignedTimedWindow returns a new TimedWindow for the given start time, end time, slot and keys.
+// We track the keys for Unaligned windows. Because for unaligned windows the start and end times for
+// each window are only dependent on the specific key.
+func NewUnalignedTimedWindow(st time.Time, et time.Time, slot string, keys []string) TimedWindow {
+	return &unalignedTimedWindow{
+		startTime: st,
+		endTime:   et,
+		slot:      slot,
 		keys:      keys,
+		partition: &SharedUnalignedPartition,
+		id:        fmt.Sprintf("%d-%d-%s-%s", st.UnixMilli(), et.UnixMilli(), slot, strings.Join(keys, "-")),
 	}
 }
 
-func (w *timedWindow) StartTime() time.Time {
+func (w *unalignedTimedWindow) StartTime() time.Time {
 	return w.startTime
 }
 
-func (w *timedWindow) EndTime() time.Time {
+func (w *unalignedTimedWindow) EndTime() time.Time {
 	return w.endTime
 }
 
-func (w *timedWindow) Slot() string {
+func (w *unalignedTimedWindow) Slot() string {
 	return w.slot
 }
 
-func (w *timedWindow) Keys() []string {
+func (w *unalignedTimedWindow) Keys() []string {
 	return w.keys
 }
 
-func (w *timedWindow) Partition() *partition.ID {
-	return &partition.ID{
-		Start: w.startTime,
-		End:   w.endTime,
-		Slot:  w.slot,
-	}
+func (w *unalignedTimedWindow) Partition() *partition.ID {
+	return w.partition
 }
 
-func (w *timedWindow) Merge(tw TimedWindow) {
+func (w *unalignedTimedWindow) ID() string {
+	return w.id
+}
+
+func (w *unalignedTimedWindow) Merge(tw TimedWindow) {
+
+	// if the window falls within the current window, no need to merge
+	if !tw.StartTime().Before(w.startTime) && !tw.EndTime().After(w.endTime) {
+		return
+	}
+
 	// expand the start and end to accommodate the new window
 	if tw.StartTime().Before(w.startTime) {
 		w.startTime = tw.StartTime()
@@ -131,11 +205,27 @@ func (w *timedWindow) Merge(tw TimedWindow) {
 	if tw.EndTime().After(w.endTime) {
 		w.endTime = tw.EndTime()
 	}
+
+	// update the id since the start and end time has changed
+	if len(w.keys) > 0 {
+		w.id = fmt.Sprintf("%d-%d-%s-%s", w.startTime.UnixMilli(), w.endTime.UnixMilli(), w.slot, strings.Join(w.keys, "-"))
+	} else {
+		w.id = fmt.Sprintf("%d-%d-%s", w.startTime.UnixMilli(), w.endTime.UnixMilli(), w.slot)
+	}
 }
 
-func (w *timedWindow) Expand(endTime time.Time) {
-	if endTime.After(w.endTime) {
-		w.endTime = endTime
+func (w *unalignedTimedWindow) Expand(endTime time.Time) {
+	// if the end time is before the current end time, no need to expand
+	if !endTime.After(w.endTime) {
+		return
+	}
+
+	w.endTime = endTime
+	// update the id since the end time has changed
+	if len(w.keys) > 0 {
+		w.id = fmt.Sprintf("%d-%d-%s-%s", w.startTime.UnixMilli(), w.endTime.UnixMilli(), w.slot, strings.Join(w.keys, "-"))
+	} else {
+		w.id = fmt.Sprintf("%d-%d-%s", w.startTime.UnixMilli(), w.endTime.UnixMilli(), w.slot)
 	}
 }
 


Test patch:
diff --git a/pkg/reduce/data_forward_test.go b/pkg/reduce/data_forward_test.go
index d0362611..6eed600e 100644
--- a/pkg/reduce/data_forward_test.go
+++ b/pkg/reduce/data_forward_test.go
@@ -165,7 +165,7 @@ func (f CounterReduceTest) ApplyReduce(_ context.Context, partitionID *partition
 			Message: ret,
 			Tags:    nil,
 		},
-		Window: window.NewWindowFromPartition(partitionID),
+		Window: window.NewAlignedTimedWindow(partitionID.Start, partitionID.End, partitionID.Slot),
 	}
 	responseCh <- response
 	return responseCh, errCh
@@ -221,7 +221,7 @@ func (s SessionSumReduceTest) ApplyReduce(ctx context.Context, partitionID *part
 					}
 					response := &window.TimedWindowResponse{
 						WriteMessage: outputMsg,
-						Window:       window.NewWindowFromPartition(partitionID),
+						Window:       window.NewAlignedTimedWindow(partitionID.Start, partitionID.End, partitionID.Slot),
 					}
 					responseCh <- response
 				}
@@ -283,7 +283,7 @@ func (s SumReduceTest) ApplyReduce(_ context.Context, partitionID *partition.ID,
 		}
 		response := &window.TimedWindowResponse{
 			WriteMessage: msg,
-			Window:       window.NewWindowFromPartition(partitionID),
+			Window:       window.NewAlignedTimedWindow(partitionID.Start, partitionID.End, partitionID.Slot),
 		}
 		responseCh <- response
 	}
@@ -342,7 +342,7 @@ func (m MaxReduceTest) ApplyReduce(_ context.Context, partitionID *partition.ID,
 
 		response := &window.TimedWindowResponse{
 			WriteMessage: ret,
-			Window:       window.NewWindowFromPartition(partitionID),
+			Window:       window.NewAlignedTimedWindow(partitionID.Start, partitionID.End, partitionID.Slot),
 		}
 		responseCh <- response
 	}
diff --git a/pkg/reduce/pnf/processandforward_test.go b/pkg/reduce/pnf/processandforward_test.go
index 3d3cb2d1..f8fa0ee8 100644
--- a/pkg/reduce/pnf/processandforward_test.go
+++ b/pkg/reduce/pnf/processandforward_test.go
@@ -24,7 +24,6 @@ import (
 
 	"github.com/stretchr/testify/assert"
 
-	"github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaflow/pkg/forwarder"
 	"github.com/numaproj/numaflow/pkg/isb"
@@ -98,15 +97,15 @@ func TestWriteToBuffer(t *testing.T) {
 	}{
 		{
 			name: "test-discard-latest",
-			buffers: []isb.BufferWriter{simplebuffer.NewInMemoryBuffer("buffer1-1", 10, 0, simplebuffer.WithBufferFullWritingStrategy(v1alpha1.DiscardLatest)),
-				simplebuffer.NewInMemoryBuffer("buffer1-2", 10, 1, simplebuffer.WithBufferFullWritingStrategy(v1alpha1.DiscardLatest))},
+			buffers: []isb.BufferWriter{simplebuffer.NewInMemoryBuffer("buffer1-1", 10, 0, simplebuffer.WithBufferFullWritingStrategy(dfv1.DiscardLatest)),
+				simplebuffer.NewInMemoryBuffer("buffer1-2", 10, 1, simplebuffer.WithBufferFullWritingStrategy(dfv1.DiscardLatest))},
 			// should not throw any error as we drop messages and finish writing before context is cancelled
 			throwError: false,
 		},
 		{
 			name: "test-retry-until-success",
-			buffers: []isb.BufferWriter{simplebuffer.NewInMemoryBuffer("buffer2-1", 10, 0, simplebuffer.WithBufferFullWritingStrategy(v1alpha1.RetryUntilSuccess)),
-				simplebuffer.NewInMemoryBuffer("buffer2-2", 10, 0, simplebuffer.WithBufferFullWritingStrategy(v1alpha1.RetryUntilSuccess))},
+			buffers: []isb.BufferWriter{simplebuffer.NewInMemoryBuffer("buffer2-1", 10, 0, simplebuffer.WithBufferFullWritingStrategy(dfv1.RetryUntilSuccess)),
+				simplebuffer.NewInMemoryBuffer("buffer2-2", 10, 0, simplebuffer.WithBufferFullWritingStrategy(dfv1.RetryUntilSuccess))},
 			// should throw context closed error as we keep retrying writing until context is cancelled
 			throwError: true,
 		},
@@ -174,7 +173,7 @@ func TestPnFHandleAlignedWindowResponses(t *testing.T) {
 	}()
 
 	windower := fixed.NewWindower(60*time.Second, keyedVertex)
-	windower.InsertWindow(window.NewWindowFromPartition(id))
+	windower.InsertWindow(window.NewAlignedTimedWindow(id.Start, id.End, id.Slot))
 
 	wmPublishers, _ := buildPublisherMapAndOTStore(toBuffersMap)
 	latestWriteOffsets := make(map[string][][]isb.Offset)
@@ -216,7 +215,7 @@ func generateAlignedWindowResponses(count int, id *partition.ID) []*window.Timed
 	for i := 0; i <= count; i++ {
 		if i == count {
 			windowResponses = append(windowResponses, &window.TimedWindowResponse{
-				Window: window.NewWindowFromPartition(id),
+				Window: window.NewAlignedTimedWindow(id.Start, id.End, id.Slot),
 				EOF:    true,
 			})
 			continue
@@ -234,7 +233,7 @@ func generateAlignedWindowResponses(count int, id *partition.ID) []*window.Timed
 					Body: isb.Body{Payload: []byte("test")},
 				},
 			},
-			Window: window.NewWindowFromPartition(id),
+			Window: window.NewAlignedTimedWindow(id.Start, id.End, id.Slot),
 			EOF:    false,
 		})
 	}
diff --git a/pkg/window/sorted_window_endtime_test.go b/pkg/window/sorted_window_endtime_test.go
index f8f77274..bcbc120e 100644
--- a/pkg/window/sorted_window_endtime_test.go
+++ b/pkg/window/sorted_window_endtime_test.go
@@ -17,12 +17,14 @@ limitations under the License.
 package window
 
 import (
+	"fmt"
+	"strings"
 	"testing"
 	"time"
 
-	"github.com/stretchr/testify/assert"
-
+	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaflow/pkg/reduce/pbq/partition"
+	"github.com/stretchr/testify/assert"
 )
 
 type TestWindow struct {
@@ -30,6 +32,7 @@ type TestWindow struct {
 	end   time.Time
 	slot  string
 	keys  []string
+	id    string
 }
 
 func (t *TestWindow) StartTime() time.Time {
@@ -53,19 +56,46 @@ func (t *TestWindow) Partition() *partition.ID {
 }
 
 func (t *TestWindow) Merge(tw TimedWindow) {
-	//TODO implement me
-	panic("implement me")
+	if tw.StartTime().Before(t.start) {
+		t.start = tw.StartTime()
+	}
+
+	if tw.EndTime().After(t.end) {
+		t.end = tw.EndTime()
+	}
+
+	if len(t.keys) > 0 {
+		t.id = fmt.Sprintf("%d-%d-%s-%s", t.start.UnixMilli(), t.end.UnixMilli(), t.slot, strings.Join(t.keys, dfv1.KeysDelimitter))
+	} else {
+		t.id = fmt.Sprintf("%d-%d-%s", t.start.UnixMilli(), t.end.UnixMilli(), t.slot)
+	}
+
 }
 
 func (t *TestWindow) Expand(et time.Time) {
-	//TODO implement me
-	panic("implement me")
+	if et.After(t.end) {
+		t.end = et
+	}
+
+	if len(t.keys) > 0 {
+		t.id = fmt.Sprintf("%d-%d-%s-%s", t.start.UnixMilli(), t.end.UnixMilli(), t.slot, strings.Join(t.keys, dfv1.KeysDelimitter))
+	} else {
+		t.id = fmt.Sprintf("%d-%d-%s", t.start.UnixMilli(), t.end.UnixMilli(), t.slot)
+	}
 }
 
 func (t *TestWindow) Keys() []string {
 	return t.keys
 }
 
+func (t *TestWindow) ID() string {
+	if len(t.id) == 0 {
+		return fmt.Sprintf("%d-%d-%s-%s", t.start.UnixMilli(), t.end.UnixMilli(), t.slot, strings.Join(t.keys, dfv1.KeysDelimitter))
+	}
+
+	return t.id
+}
+
 func TestSortedWindowListByEndTime_InsertIfNotPresent(t *testing.T) {
 	tests := []struct {
 		name            string
diff --git a/pkg/window/strategy/fixed/fixed_test.go b/pkg/window/strategy/fixed/fixed_test.go
index 18d000ff..796779a6 100644
--- a/pkg/window/strategy/fixed/fixed_test.go
+++ b/pkg/window/strategy/fixed/fixed_test.go
@@ -16,6 +16,7 @@ limitations under the License.
 package fixed
 
 import (
+	"fmt"
 	"testing"
 	"time"
 
@@ -153,14 +154,34 @@ func TestFixed_DeleteWindows(t *testing.T) {
 		startTime: baseTime,
 		endTime:   baseTime.Add(60 * time.Second),
 		slot:      "slot-0",
+		partition: &partition.ID{
+			Start: baseTime,
+			End:   baseTime.Add(60 * time.Second),
+			Slot:  "slot-0",
+		},
+		id: fmt.Sprintf("%d-%d-%s", baseTime.UnixMilli(), baseTime.Add(60*time.Second).UnixMilli(), "slot-0"),
 	}
 	win2 := &fixedWindow{
 		startTime: baseTime.Add(60 * time.Second),
 		endTime:   baseTime.Add(120 * time.Second),
+		slot:      "slot-0",
+		partition: &partition.ID{
+			Start: baseTime.Add(60 * time.Second),
+			End:   baseTime.Add(120 * time.Second),
+			Slot:  "slot-0",
+		},
+		id: fmt.Sprintf("%d-%d-%s", baseTime.Add(60*time.Second).UnixMilli(), baseTime.Add(120*time.Second).UnixMilli(), "slot-0"),
 	}
 	win3 := &fixedWindow{
 		startTime: baseTime.Add(120 * time.Second),
 		endTime:   baseTime.Add(180 * time.Second),
+		slot:      "slot-0",
+		partition: &partition.ID{
+			Start: baseTime.Add(120 * time.Second),
+			End:   baseTime.Add(180 * time.Second),
+			Slot:  "slot-0",
+		},
+		id: fmt.Sprintf("%d-%d-%s", baseTime.Add(120*time.Second).UnixMilli(), baseTime.Add(180*time.Second).UnixMilli(), "slot-0"),
 	}
 
 	windower := &Windower{
@@ -178,11 +199,17 @@ func TestFixed_DeleteWindows(t *testing.T) {
 	windower.CloseWindows(baseTime.Add(180 * time.Second))
 
 	// delete one of the windows
-	windower.DeleteClosedWindow(window.NewWindowFromPartition(&partition.ID{
-		Start: baseTime,
-		End:   baseTime.Add(60 * time.Second),
-		Slot:  "slot-0",
-	}))
+	windower.DeleteClosedWindow(&fixedWindow{
+		startTime: baseTime.Add(60 * time.Second),
+		endTime:   baseTime.Add(120 * time.Second),
+		slot:      "slot-0",
+		partition: &partition.ID{
+			Start: baseTime.Add(60 * time.Second),
+			End:   baseTime.Add(120 * time.Second),
+			Slot:  "slot-0",
+		},
+		id: fmt.Sprintf("%d-%d-%s", baseTime.Add(60*time.Second).UnixMilli(), baseTime.Add(120*time.Second).UnixMilli(), "slot-0"),
+	})
 
 	// since we deleted one of the windows, the closed windows should be 2
 	assert.Equal(t, 2, windower.closedWindows.Len())
@@ -195,14 +222,34 @@ func TestFixed_OldestClosedWindowEndTime(t *testing.T) {
 		startTime: baseTime,
 		endTime:   baseTime.Add(60 * time.Second),
 		slot:      "slot-0",
+		partition: &partition.ID{
+			Start: baseTime,
+			End:   baseTime.Add(60 * time.Second),
+			Slot:  "slot-0",
+		},
+		id: fmt.Sprintf("%d-%d-%s", baseTime.UnixMilli(), baseTime.Add(60*time.Second).UnixMilli(), "slot-0"),
 	}
 	win2 := &fixedWindow{
 		startTime: baseTime.Add(60 * time.Second),
 		endTime:   baseTime.Add(120 * time.Second),
+		slot:      "slot-0",
+		partition: &partition.ID{
+			Start: baseTime.Add(60 * time.Second),
+			End:   baseTime.Add(120 * time.Second),
+			Slot:  "slot-0",
+		},
+		id: fmt.Sprintf("%d-%d-%s", baseTime.Add(60*time.Second).UnixMilli(), baseTime.Add(120*time.Second).UnixMilli(), "slot-0"),
 	}
 	win3 := &fixedWindow{
 		startTime: baseTime.Add(120 * time.Second),
 		endTime:   baseTime.Add(180 * time.Second),
+		slot:      "slot-0",
+		partition: &partition.ID{
+			Start: baseTime.Add(120 * time.Second),
+			End:   baseTime.Add(180 * time.Second),
+			Slot:  "slot-0",
+		},
+		id: fmt.Sprintf("%d-%d-%s", baseTime.Add(120*time.Second).UnixMilli(), baseTime.Add(180*time.Second).UnixMilli(), "slot-0"),
 	}
 
 	windower := &Windower{
@@ -230,14 +277,34 @@ func TestWindower_NextWindowToBeClosed(t *testing.T) {
 		startTime: baseTime,
 		endTime:   baseTime.Add(60 * time.Second),
 		slot:      "slot-0",
+		partition: &partition.ID{
+			Start: baseTime,
+			End:   baseTime.Add(60 * time.Second),
+			Slot:  "slot-0",
+		},
+		id: fmt.Sprintf("%d-%d-%s", baseTime.UnixMilli(), baseTime.Add(60*time.Second).UnixMilli(), "slot-0"),
 	}
 	win2 := &fixedWindow{
 		startTime: baseTime.Add(60 * time.Second),
 		endTime:   baseTime.Add(120 * time.Second),
+		slot:      "slot-0",
+		partition: &partition.ID{
+			Start: baseTime.Add(60 * time.Second),
+			End:   baseTime.Add(120 * time.Second),
+			Slot:  "slot-0",
+		},
+		id: fmt.Sprintf("%d-%d-%s", baseTime.Add(60*time.Second).UnixMilli(), baseTime.Add(120*time.Second).UnixMilli(), "slot-0"),
 	}
 	win3 := &fixedWindow{
 		startTime: baseTime.Add(120 * time.Second),
 		endTime:   baseTime.Add(180 * time.Second),
+		slot:      "slot-0",
+		partition: &partition.ID{
+			Start: baseTime.Add(120 * time.Second),
+			End:   baseTime.Add(180 * time.Second),
+			Slot:  "slot-0",
+		},
+		id: fmt.Sprintf("%d-%d-%s", baseTime.Add(120*time.Second).UnixMilli(), baseTime.Add(180*time.Second).UnixMilli(), "slot-0"),
 	}
 
 	windower := &Windower{
diff --git a/pkg/window/strategy/session/session_test.go b/pkg/window/strategy/session/session_test.go
index b4b81360..d191ce96 100644
--- a/pkg/window/strategy/session/session_test.go
+++ b/pkg/window/strategy/session/session_test.go
@@ -16,6 +16,7 @@ limitations under the License.
 package session
 
 import (
+	"fmt"
 	"math"
 	"testing"
 	"time"
@@ -24,7 +25,6 @@ import (
 
 	dfv1 "github.com/numaproj/numaflow/pkg/apis/numaflow/v1alpha1"
 	"github.com/numaproj/numaflow/pkg/isb"
-	"github.com/numaproj/numaflow/pkg/reduce/pbq/partition"
 	"github.com/numaproj/numaflow/pkg/window"
 )
 
@@ -67,7 +67,7 @@ func TestAssignWindows_NewWindow(t *testing.T) {
 	assert.Equal(t, 1, len(windowOperations[0].Windows))
 	assert.Equal(t, baseTime, windowOperations[0].Windows[0].StartTime())
 	assert.Equal(t, baseTime.Add(gap), windowOperations[0].Windows[0].EndTime())
-	assert.Equal(t, &SharedSessionPartition, windowOperations[0].ID)
+	assert.Equal(t, &window.SharedUnalignedPartition, windowOperations[0].ID)
 
 	// 2nd message should be assigned to the same window, since key is same
 	message = buildReadMessage(baseTime.Add(5*time.Second), []string{"key1"})
@@ -81,7 +81,7 @@ func TestAssignWindows_NewWindow(t *testing.T) {
 	// 0th index we should have the old window, and 1st index we should have the new window
 	assert.Equal(t, baseTime.Add(gap), windowOperations[0].Windows[0].EndTime())
 	assert.Equal(t, message.EventTime.Add(gap), windowOperations[0].Windows[1].EndTime())
-	assert.Equal(t, &SharedSessionPartition, windowOperations[0].ID)
+	assert.Equal(t, &window.SharedUnalignedPartition, windowOperations[0].ID)
 
 	// 3rd message should be assigned to a new window, since key is different
 	message = buildReadMessage(baseTime, []string{"key2"})
@@ -93,7 +93,7 @@ func TestAssignWindows_NewWindow(t *testing.T) {
 	assert.Equal(t, 1, len(windowOperations[0].Windows))
 	assert.Equal(t, baseTime, windowOperations[0].Windows[0].StartTime())
 	assert.Equal(t, baseTime.Add(gap), windowOperations[0].Windows[0].EndTime())
-	assert.Equal(t, &SharedSessionPartition, windowOperations[0].ID)
+	assert.Equal(t, &window.SharedUnalignedPartition, windowOperations[0].ID)
 
 	// 4th message should be assigned to a new window, because of the gap duration
 	message = buildReadMessage(baseTime.Add(20*time.Second), []string{"key2"})
@@ -105,22 +105,27 @@ func TestAssignWindows_NewWindow(t *testing.T) {
 	assert.Equal(t, 1, len(windowOperations[0].Windows))
 	assert.Equal(t, baseTime.Add(20*time.Second), windowOperations[0].Windows[0].StartTime())
 	assert.Equal(t, baseTime.Add(30*time.Second), windowOperations[0].Windows[0].EndTime())
-	assert.Equal(t, &SharedSessionPartition, windowOperations[0].ID)
+	assert.Equal(t, &window.SharedUnalignedPartition, windowOperations[0].ID)
 }
 
 func TestSession_InsertWindow(t *testing.T) {
+	baseTime := time.UnixMilli(60000)
 	win1 := &sessionWindow{
-		startTime: time.UnixMilli(60000),
-		endTime:   time.UnixMilli(60000 + 60*1000),
+		startTime: baseTime,
+		endTime:   baseTime.Add(60 * time.Second),
 		slot:      "slot-0",
 		keys:      []string{"key-1"},
+		partition: &window.SharedUnalignedPartition,
+		id:        fmt.Sprintf("%d-%d-%s-%s", baseTime.UnixMilli(), baseTime.Add(60*time.Second).UnixMilli(), "slot-0", "key-1"),
 	}
 
 	win2 := &sessionWindow{
-		startTime: time.UnixMilli(60000),
-		endTime:   time.UnixMilli(60000 + 60*1000),
+		startTime: baseTime,
+		endTime:   baseTime.Add(60 * time.Second),
 		slot:      "slot-0",
 		keys:      []string{"key-2"},
+		partition: &window.SharedUnalignedPartition,
+		id:        fmt.Sprintf("%d-%d-%s-%s", baseTime.UnixMilli(), baseTime.Add(60*time.Second).UnixMilli(), "slot-0", "key-2"),
 	}
 
 	windower := &Windower{
@@ -139,6 +144,8 @@ func TestSession_InsertWindow(t *testing.T) {
 		endTime:   time.UnixMilli(120000 + 60*1000),
 		slot:      "slot-0",
 		keys:      []string{"key-2"},
+		partition: &window.SharedUnalignedPartition,
+		id:        fmt.Sprintf("%d-%d-%s-%s", time.UnixMilli(120000).UnixMilli(), time.UnixMilli(120000+60*1000).UnixMilli(), "slot-0", "key-2"),
 	}
 
 	windower.InsertWindow(win2)
@@ -154,14 +161,25 @@ func TestSession_CloseWindowsWithoutMerge(t *testing.T) {
 		startTime: baseTime,
 		endTime:   baseTime.Add(10 * time.Second),
 		slot:      "slot-0",
+		partition: &window.SharedUnalignedPartition,
+		keys:      []string{"key-1"},
+		id:        fmt.Sprintf("%d-%d-%s-%s", baseTime.UnixMilli(), baseTime.Add(10*time.Second).UnixMilli(), "slot-0", "key-1"),
 	}
 	win2 := &sessionWindow{
 		startTime: baseTime.Add(30 * time.Second),
 		endTime:   baseTime.Add(59 * time.Second),
+		slot:      "slot-0",
+		partition: &window.SharedUnalignedPartition,
+		keys:      []string{"key-1"},
+		id:        fmt.Sprintf("%d-%d-%s-%s", baseTime.Add(30*time.Second).UnixMilli(), baseTime.Add(59*time.Second).UnixMilli(), "slot-0", "key-1"),
 	}
 	win3 := &sessionWindow{
 		startTime: baseTime.Add(60 * time.Second),
 		endTime:   baseTime.Add(90 * time.Second),
+		slot:      "slot-0",
+		partition: &window.SharedUnalignedPartition,
+		keys:      []string{"key-1"},
+		id:        fmt.Sprintf("%d-%d-%s-%s", baseTime.Add(60*time.Second).UnixMilli(), baseTime.Add(90*time.Second).UnixMilli(), "slot-0", "key-1"),
 	}
 
 	windower := NewWindower(10*time.Second, keyedVertex)
@@ -199,16 +217,25 @@ func TestSession_CloseWindowsWithMerge(t *testing.T) {
 		startTime: baseTime,
 		endTime:   baseTime.Add(20 * time.Second),
 		slot:      "slot-0",
+		partition: &window.SharedUnalignedPartition,
+		keys:      []string{"key-1"},
+		id:        fmt.Sprintf("%d-%d-%s-%s", baseTime.UnixMilli(), baseTime.Add(20*time.Second).UnixMilli(), "slot-0", "key-1"),
 	}
 	win2 := &sessionWindow{
 		startTime: baseTime.Add(11 * time.Second),
 		endTime:   baseTime.Add(59 * time.Second),
 		slot:      "slot-0",
+		partition: &window.SharedUnalignedPartition,
+		keys:      []string{"key-1"},
+		id:        fmt.Sprintf("%d-%d-%s-%s", baseTime.Add(11*time.Second).UnixMilli(), baseTime.Add(59*time.Second).UnixMilli(), "slot-0", "key-1"),
 	}
 	win3 := &sessionWindow{
 		startTime: baseTime.Add(60 * time.Second),
 		endTime:   baseTime.Add(90 * time.Second),
 		slot:      "slot-0",
+		partition: &window.SharedUnalignedPartition,
+		keys:      []string{"key-1"},
+		id:        fmt.Sprintf("%d-%d-%s-%s", baseTime.Add(60*time.Second).UnixMilli(), baseTime.Add(90*time.Second).UnixMilli(), "slot-0", "key-1"),
 	}
 
 	windower := NewWindower(10*time.Second, keyedVertex)
@@ -246,18 +273,24 @@ func TestSession_CloseWindowsWithMerge(t *testing.T) {
 		endTime:   baseTime.Add(89 * time.Second),
 		slot:      "slot-0",
 		keys:      []string{"key-1"},
+		partition: &window.SharedUnalignedPartition,
+		id:        fmt.Sprintf("%d-%d-%s-%s", baseTime.Add(60*time.Second).UnixMilli(), baseTime.Add(89*time.Second).UnixMilli(), "slot-0", "key-1"),
 	}
 	win2 = &sessionWindow{
 		startTime: baseTime.Add(50 * time.Second),
 		endTime:   baseTime.Add(120 * time.Second),
 		slot:      "slot-0",
 		keys:      []string{"key-1"},
+		partition: &window.SharedUnalignedPartition,
+		id:        fmt.Sprintf("%d-%d-%s-%s", baseTime.Add(50*time.Second).UnixMilli(), baseTime.Add(120*time.Second).UnixMilli(), "slot-0", "key-1"),
 	}
 	win3 = &sessionWindow{
 		startTime: baseTime.Add(30 * time.Second),
 		endTime:   baseTime.Add(150 * time.Second),
 		slot:      "slot-0",
 		keys:      []string{"key-1"},
+		partition: &window.SharedUnalignedPartition,
+		id:        fmt.Sprintf("%d-%d-%s-%s", baseTime.Add(30*time.Second).UnixMilli(), baseTime.Add(150*time.Second).UnixMilli(), "slot-0", "key-1"),
 	}
 
 	win4 := &sessionWindow{
@@ -265,18 +298,24 @@ func TestSession_CloseWindowsWithMerge(t *testing.T) {
 		endTime:   baseTime.Add(50 * time.Second),
 		slot:      "slot-0",
 		keys:      []string{"key-2"},
+		partition: &window.SharedUnalignedPartition,
+		id:        fmt.Sprintf("%d-%d-%s-%s", baseTime.Add(30*time.Second).UnixMilli(), baseTime.Add(50*time.Second).UnixMilli(), "slot-0", "key-2"),
 	}
 	win5 := &sessionWindow{
 		startTime: baseTime.Add(40 * time.Second),
 		endTime:   baseTime.Add(80 * time.Second),
 		slot:      "slot-0",
 		keys:      []string{"key-2"},
+		partition: &window.SharedUnalignedPartition,
+		id:        fmt.Sprintf("%d-%d-%s-%s", baseTime.Add(40*time.Second).UnixMilli(), baseTime.Add(80*time.Second).UnixMilli(), "slot-0", "key-2"),
 	}
 	win6 := &sessionWindow{
 		startTime: baseTime.Add(10 * time.Second),
 		endTime:   baseTime.Add(90 * time.Second),
 		slot:      "slot-0",
 		keys:      []string{"key-2"},
+		partition: &window.SharedUnalignedPartition,
+		id:        fmt.Sprintf("%d-%d-%s-%s", baseTime.Add(10*time.Second).UnixMilli(), baseTime.Add(90*time.Second).UnixMilli(), "slot-0", "key-2"),
 	}
 
 	windower.InsertWindow(win1)
@@ -318,6 +357,8 @@ func TestSession_CloseWindowsComplex(t *testing.T) {
 		endTime:   time.Unix(119, 0),
 		slot:      "slot-0",
 		keys:      []string{"key-1"},
+		partition: &window.SharedUnalignedPartition,
+		id:        fmt.Sprintf("%d-%d-%s-%s", 60, 119, "slot-0", "key-1"),
 	}
 
 	win2 := &sessionWindow{
@@ -325,6 +366,8 @@ func TestSession_CloseWindowsComplex(t *testing.T) {
 		endTime:   time.Unix(218, 0),
 		slot:      "slot-0",
 		keys:      []string{"key-1"},
+		partition: &window.SharedUnalignedPartition,
+		id:        fmt.Sprintf("%d-%d-%s-%s", 159, 218, "slot-0", "key-1"),
 	}
 
 	win3 := &sessionWindow{
@@ -332,6 +375,8 @@ func TestSession_CloseWindowsComplex(t *testing.T) {
 		endTime:   time.Unix(279, 0),
 		slot:      "slot-0",
 		keys:      []string{"key-1"},
+		partition: &window.SharedUnalignedPartition,
+		id:        fmt.Sprintf("%d-%d-%s-%s", 258, 279, "slot-0", "key-1"),
 	}
 
 	win4 := &sessionWindow{
@@ -339,6 +384,8 @@ func TestSession_CloseWindowsComplex(t *testing.T) {
 		endTime:   time.Unix(317, 0),
 		slot:      "slot-0",
 		keys:      []string{"key-1"},
+		partition: &window.SharedUnalignedPartition,
+		id:        fmt.Sprintf("%d-%d-%s-%s", 262, 317, "slot-0", "key-1"),
 	}
 
 	win5 := &sessionWindow{
@@ -346,6 +393,8 @@ func TestSession_CloseWindowsComplex(t *testing.T) {
 		endTime:   time.Unix(416, 0),
 		slot:      "slot-0",
 		keys:      []string{"key-1"},
+		partition: &window.SharedUnalignedPartition,
+		id:        fmt.Sprintf("%d-%d-%s-%s", 357, 416, "slot-0", "key-1"),
 	}
 
 	win6 := &sessionWindow{
@@ -353,6 +402,8 @@ func TestSession_CloseWindowsComplex(t *testing.T) {
 		endTime:   time.Unix(515, 0),
 		slot:      "slot-0",
 		keys:      []string{"key-1"},
+		partition: &window.SharedUnalignedPartition,
+		id:        fmt.Sprintf("%d-%d-%s-%s", 456, 515, "slot-0", "key-1"),
 	}
 
 	win7 := &sessionWindow{
@@ -360,6 +411,8 @@ func TestSession_CloseWindowsComplex(t *testing.T) {
 		endTime:   time.Unix(573, 0),
 		slot:      "slot-0",
 		keys:      []string{"key-1"},
+		partition: &window.SharedUnalignedPartition,
+		id:        fmt.Sprintf("%d-%d-%s-%s", 514, 573, "slot-0", "key-1"),
 	}
 
 	windower := &Windower{
@@ -434,24 +487,32 @@ func TestSession_DeleteWindows(t *testing.T) {
 		endTime:   baseTime.Add(60 * time.Second),
 		slot:      "slot-0",
 		keys:      []string{"key-1"},
+		partition: &window.SharedUnalignedPartition,
+		id:        fmt.Sprintf("%d-%d-%s-%s", baseTime.UnixMilli(), baseTime.Add(60*time.Second).UnixMilli(), "slot-0", "key-1"),
 	}
 	win2 := &sessionWindow{
 		startTime: baseTime.Add(60 * time.Second),
 		endTime:   baseTime.Add(120 * time.Second),
 		slot:      "slot-0",
 		keys:      []string{"key-1"},
+		partition: &window.SharedUnalignedPartition,
+		id:        fmt.Sprintf("%d-%d-%s-%s", baseTime.Add(60*time.Second).UnixMilli(), baseTime.Add(120*time.Second).UnixMilli(), "slot-0", "key-1"),
 	}
 	win3 := &sessionWindow{
 		startTime: baseTime.Add(60 * time.Second),
 		endTime:   baseTime.Add(120 * time.Second),
 		slot:      "slot-0",
 		keys:      []string{"key-2"},
+		partition: &window.SharedUnalignedPartition,
+		id:        fmt.Sprintf("%d-%d-%s-%s", baseTime.Add(60*time.Second).UnixMilli(), baseTime.Add(120*time.Second).UnixMilli(), "slot-0", "key-2"),
 	}
 	win4 := &sessionWindow{
 		startTime: baseTime.Add(120 * time.Second),
 		endTime:   baseTime.Add(180 * time.Second),
 		slot:      "slot-0",
 		keys:      []string{"key-2"},
+		partition: &window.SharedUnalignedPartition,
+		id:        fmt.Sprintf("%d-%d-%s-%s", baseTime.Add(120*time.Second).UnixMilli(), baseTime.Add(180*time.Second).UnixMilli(), "slot-0", "key-2"),
 	}
 
 	windower := &Windower{
@@ -470,20 +531,22 @@ func TestSession_DeleteWindows(t *testing.T) {
 	windower.CloseWindows(baseTime.Add(180 * time.Second))
 
 	// delete one of the windows
-	windower.DeleteClosedWindow(window.NewWindowFromPartitionAndKeys(&partition.ID{
-		Start: baseTime,
-		End:   baseTime.Add(60 * time.Second),
-		Slot:  "slot-0",
-	}, []string{"key-1"}))
+	windower.DeleteClosedWindow(window.NewUnalignedTimedWindow(
+		baseTime,
+		baseTime.Add(60*time.Second),
+		"slot-0",
+		[]string{"key-1"},
+	))
 
 	// since we deleted one of the windows, the closed windows should be 3
 	assert.Equal(t, 3, windower.closedWindows.Len())
 
-	windower.DeleteClosedWindow(window.NewWindowFromPartitionAndKeys(&partition.ID{
-		Start: baseTime.Add(60 * time.Second),
-		End:   baseTime.Add(120 * time.Second),
-		Slot:  "slot-0",
-	}, []string{"key-1"}))
+	windower.DeleteClosedWindow(window.NewUnalignedTimedWindow(
+		baseTime.Add(60*time.Second),
+		baseTime.Add(120*time.Second),
+		"slot-0",
+		[]string{"key-1"},
+	))
 	// since we deleted two windows, the closed windows should be 2
 	assert.Equal(t, 2, windower.closedWindows.Len())
 }
@@ -496,18 +559,24 @@ func TestWindower_OldestClosedWindowEndTime(t *testing.T) {
 		endTime:   baseTime.Add(60 * time.Second),
 		slot:      "slot-0",
 		keys:      []string{"key-1"},
+		partition: &window.SharedUnalignedPartition,
+		id:        fmt.Sprintf("%d-%d-%s-%s", baseTime.UnixMilli(), baseTime.Add(60*time.Second).UnixMilli(), "slot-0", "key-1"),
 	}
 	win2 := &sessionWindow{
 		startTime: baseTime.Add(40 * time.Second),
 		endTime:   baseTime.Add(70 * time.Second),
 		slot:      "slot-0",
 		keys:      []string{"key-1"},
+		partition: &window.SharedUnalignedPartition,
+		id:        fmt.Sprintf("%d-%d-%s-%s", baseTime.Add(40*time.Second).UnixMilli(), baseTime.Add(70*time.Second).UnixMilli(), "slot-0", "key-1"),
 	}
 	win3 := &sessionWindow{
 		startTime: baseTime.Add(10 * time.Second),
 		endTime:   baseTime.Add(90 * time.Second),
 		slot:      "slot-0",
 		keys:      []string{"key-2"},
+		partition: &window.SharedUnalignedPartition,
+		id:        fmt.Sprintf("%d-%d-%s-%s", baseTime.Add(10*time.Second).UnixMilli(), baseTime.Add(90*time.Second).UnixMilli(), "slot-0", "key-2"),
 	}
 
 	windower := &Windower{
@@ -529,11 +598,12 @@ func TestWindower_OldestClosedWindowEndTime(t *testing.T) {
 	assert.Equal(t, baseTime.Add(70*time.Second), windower.OldestWindowEndTime())
 
 	// delete one of the windows
-	windower.DeleteClosedWindow(window.NewWindowFromPartitionAndKeys(&partition.ID{
-		Start: baseTime,
-		End:   baseTime.Add(70 * time.Second),
-		Slot:  "slot-0",
-	}, []string{"key-1"}))
+	windower.DeleteClosedWindow(window.NewUnalignedTimedWindow(
+		baseTime,
+		baseTime.Add(70*time.Second),
+		"slot-0",
+		[]string{"key-1"},
+	))
 
 	// since we deleted (60, 130) window, now the oldest window end time will be 90
 	assert.Equal(t, baseTime.Add(90*time.Second), windower.OldestWindowEndTime())
@@ -547,18 +617,24 @@ func TestWindower_NextWindowToBeClosed(t *testing.T) {
 		endTime:   baseTime.Add(10 * time.Second),
 		slot:      "slot-0",
 		keys:      []string{"key-1"},
+		partition: &window.SharedUnalignedPartition,
+		id:        fmt.Sprintf("%d-%d-%s-%s", baseTime.UnixMilli(), baseTime.Add(10*time.Second).UnixMilli(), "slot-0", "key-1"),
 	}
 	win2 := &sessionWindow{
 		startTime: baseTime.Add(30 * time.Second),
 		endTime:   baseTime.Add(59 * time.Second),
 		slot:      "slot-0",
 		keys:      []string{"key-1"},
+		partition: &window.SharedUnalignedPartition,
+		id:        fmt.Sprintf("%d-%d-%s-%s", baseTime.Add(30*time.Second).UnixMilli(), baseTime.Add(59*time.Second).UnixMilli(), "slot-0", "key-1"),
 	}
 	win3 := &sessionWindow{
 		startTime: baseTime.Add(60 * time.Second),
 		endTime:   baseTime.Add(90 * time.Second),
 		slot:      "slot-0",
 		keys:      []string{"key-1"},
+		partition: &window.SharedUnalignedPartition,
+		id:        fmt.Sprintf("%d-%d-%s-%s", baseTime.Add(60*time.Second).UnixMilli(), baseTime.Add(90*time.Second).UnixMilli(), "slot-0", "key-1"),
 	}
 
 	windower := NewWindower(10*time.Second, keyedVertex)
diff --git a/pkg/window/strategy/sliding/sliding_test.go b/pkg/window/strategy/sliding/sliding_test.go
index 59d6e4db..0c9e10fe 100644
--- a/pkg/window/strategy/sliding/sliding_test.go
+++ b/pkg/window/strategy/sliding/sliding_test.go
@@ -16,6 +16,7 @@ limitations under the License.
 package sliding
 
 import (
+	"fmt"
 	"testing"
 	"time"
 
@@ -126,6 +127,7 @@ func TestSliding_InsertWindow(t *testing.T) {
 		startTime: time.UnixMilli(60000),
 		endTime:   time.UnixMilli(60000 + 60*1000),
 		slot:      "slot-0",
+		id:        fmt.Sprintf("%d-%d-%s", 60000, 60000+60*1000, "slot-0"),
 	}
 
 	windower := &Windower{
@@ -162,14 +164,19 @@ func TestSliding_CloseWindows(t *testing.T) {
 		startTime: baseTime,
 		endTime:   baseTime.Add(60 * time.Second),
 		slot:      "slot-0",
+		id:        fmt.Sprintf("%d-%d-%s", baseTime.UnixMilli(), baseTime.Add(60*time.Second).UnixMilli(), "slot-0"),
 	}
 	win2 := &slidingWindow{
 		startTime: baseTime.Add(-10 * time.Second),
 		endTime:   baseTime.Add(50 * time.Second),
+		slot:      "slot-0",
+		id:        fmt.Sprintf("%d-%d-%s", baseTime.Add(-10*time.Second).UnixMilli(), baseTime.Add(50*time.Second).UnixMilli(), "slot-0"),
 	}
 	win3 := &slidingWindow{
 		startTime: baseTime.Add(-20 * time.Second),
 		endTime:   baseTime.Add(40 * time.Second),
+		slot:      "slot-0",
+		id:        fmt.Sprintf("%d-%d-%s", baseTime.Add(-20*time.Second).UnixMilli(), baseTime.Add(40*time.Second).UnixMilli(), "slot-0"),
 	}
 
 	windower := NewWindower(60*time.Second, 10*time.Second, keyedVertex)
@@ -204,14 +211,19 @@ func TestSliding_DeleteWindows(t *testing.T) {
 		startTime: baseTime,
 		endTime:   baseTime.Add(60 * time.Second),
 		slot:      "slot-0",
+		id:        fmt.Sprintf("%d-%d-%s", baseTime.UnixMilli(), baseTime.Add(60*time.Second).UnixMilli(), "slot-0"),
 	}
 	win2 := &slidingWindow{
 		startTime: baseTime.Add(-10 * time.Second),
 		endTime:   baseTime.Add(50 * time.Second),
+		slot:      "slot-0",
+		id:        fmt.Sprintf("%d-%d-%s", baseTime.Add(-10*time.Second).UnixMilli(), baseTime.Add(50*time.Second).UnixMilli(), "slot-0"),
 	}
 	win3 := &slidingWindow{
 		startTime: baseTime.Add(-20 * time.Second),
 		endTime:   baseTime.Add(40 * time.Second),
+		slot:      "slot-0",
+		id:        fmt.Sprintf("%d-%d-%s", baseTime.Add(-20*time.Second).UnixMilli(), baseTime.Add(40*time.Second).UnixMilli(), "slot-0"),
 	}
 
 	windower := &Windower{
@@ -230,11 +242,17 @@ func TestSliding_DeleteWindows(t *testing.T) {
 	windower.CloseWindows(baseTime.Add(120 * time.Second))
 
 	// delete one of the windows
-	windower.DeleteClosedWindow(window.NewWindowFromPartition(&partition.ID{
-		Start: baseTime,
-		End:   baseTime.Add(60 * time.Second),
-		Slot:  "slot-0",
-	}))
+	windower.DeleteClosedWindow(&slidingWindow{
+		startTime: baseTime,
+		endTime:   baseTime.Add(60 * time.Second),
+		slot:      "slot-0",
+		id:        fmt.Sprintf("%d-%d-%s", baseTime.UnixMilli(), baseTime.Add(60*time.Second).UnixMilli(), "slot-0"),
+		partition: &partition.ID{
+			Slot:  "slot-0",
+			Start: baseTime,
+			End:   baseTime.Add(60 * time.Second),
+		},
+	})
 
 	// since we deleted one of the windows, the closed windows should be 2
 	assert.Equal(t, 2, windower.closedWindows.Len())
@@ -247,14 +265,19 @@ func TestSliding_OldestClosedWindowEndTime(t *testing.T) {
 		startTime: baseTime,
 		endTime:   baseTime.Add(60 * time.Second),
 		slot:      "slot-0",
+		id:        fmt.Sprintf("%d-%d-%s", baseTime.UnixMilli(), baseTime.Add(60*time.Second).UnixMilli(), "slot-0"),
 	}
 	win2 := &slidingWindow{
 		startTime: baseTime.Add(-10 * time.Second),
 		endTime:   baseTime.Add(50 * time.Second),
+		slot:      "slot-0",
+		id:        fmt.Sprintf("%d-%d-%s", baseTime.Add(-10*time.Second).UnixMilli(), baseTime.Add(50*time.Second).UnixMilli(), "slot-0"),
 	}
 	win3 := &slidingWindow{
 		startTime: baseTime.Add(-20 * time.Second),
 		endTime:   baseTime.Add(40 * time.Second),
+		slot:      "slot-0",
+		id:        fmt.Sprintf("%d-%d-%s", baseTime.Add(-20*time.Second).UnixMilli(), baseTime.Add(40*time.Second).UnixMilli(), "slot-0"),
 	}
 
 	windower := &Windower{
@@ -283,14 +306,19 @@ func TestSliding_NextWindowToBeClosed(t *testing.T) {
 		startTime: baseTime,
 		endTime:   baseTime.Add(60 * time.Second),
 		slot:      "slot-0",
+		id:        fmt.Sprintf("%d-%d-%s", baseTime.UnixMilli(), baseTime.Add(60*time.Second).UnixMilli(), "slot-0"),
 	}
 	win2 := &slidingWindow{
 		startTime: baseTime.Add(-10 * time.Second),
 		endTime:   baseTime.Add(50 * time.Second),
+		slot:      "slot-0",
+		id:        fmt.Sprintf("%d-%d-%s", baseTime.Add(-10*time.Second).UnixMilli(), baseTime.Add(50*time.Second).UnixMilli(), "slot-0"),
 	}
 	win3 := &slidingWindow{
 		startTime: baseTime.Add(-20 * time.Second),
 		endTime:   baseTime.Add(40 * time.Second),
+		slot:      "slot-0",
+		id:        fmt.Sprintf("%d-%d-%s", baseTime.Add(-20*time.Second).UnixMilli(), baseTime.Add(40*time.Second).UnixMilli(), "slot-0"),
 	}
 
 	windower := &Windower{