Lift the equality check from validation to simplify session windows

Author: gwaramadze

quixstreams/dataframe/windows/session.py

@@ -94,14 +94,14 @@ def process_window(
 
         for (window_start, window_end), aggregated_value, _ in windows:
             # Calculate the time gap between the new event and the session's last activity
-            # window_end is stored as last_event_timestamp + 1, so subtract 1 to get actual last event time
-            session_last_activity = window_end - 1
+            # window_end now directly represents the timestamp of the last event
+            session_last_activity = window_end
             time_gap = timestamp_ms - session_last_activity
 
             # Check if this session can be extended
             if time_gap <= timeout_ms + grace_ms and timestamp_ms >= window_start:
                 session_start = window_start
-                session_end = timestamp_ms + 1
+                session_end = timestamp_ms
                 can_extend_session = True
                 existing_aggregated = aggregated_value
                 old_window_to_delete = (window_start, window_end)
@@ -110,7 +110,7 @@ def process_window(
         # If no extendable session found, start a new one
         if not can_extend_session:
             session_start = timestamp_ms
-            session_end = timestamp_ms + 1
+            session_end = timestamp_ms  # End time is the timestamp of the last event
 
         # Process the event for this session
         updated_windows: list[WindowKeyResult] = []
@@ -138,7 +138,7 @@ def process_window(
             updated_windows.append(
                 (
                     key,
-                    self._results(aggregated, [], session_start, session_end - 1),
+                    self._results(aggregated, [], session_start, session_end),
                 )
             )
 
@@ -221,18 +221,18 @@ def expire_by_key(
         windows_to_delete = []
         for (window_start, window_end), aggregated, _ in all_windows:
             # Session expires when the session end time + timeout has passed the expiry threshold
-            # window_end is stored as last_event_timestamp + 1, so we subtract 1 and add timeout_ms
-            last_event_timestamp = window_end - 1
-            if last_event_timestamp + self._timeout_ms <= expiry_threshold:
+            # window_end directly represents the timestamp of the last event
+            if window_end + self._timeout_ms <= expiry_threshold:
                 collected = []
                 if collect:
-                    collected = state.get_from_collection(window_start, window_end)
+                    # window_end is now the timestamp of the last event, so we need +1 to include it
+                    collected = state.get_from_collection(window_start, window_end + 1)
 
                 windows_to_delete.append((window_start, window_end))
                 count += 1
                 yield (
                     key,
-                    self._results(aggregated, collected, window_start, window_end - 1),
+                    self._results(aggregated, collected, window_start, window_end),
                 )
 
         # Clean up expired windows

quixstreams/state/rocksdb/windowed/transaction.py

@@ -485,9 +485,9 @@ def _set_timestamp(self, cache: Cache, prefix: bytes, timestamp_ms: int):
         )
 
     def _validate_duration(self, start_ms: int, end_ms: int):
-        if end_ms <= start_ms:
+        if end_ms < start_ms:
             raise ValueError(
-                f"Invalid window duration: window end {end_ms} is smaller or equal "
+                f"Invalid window duration: window end {end_ms} is smaller "
                 f"than window start {start_ms}"
             )