End time equals to last event time

Author: gwaramadze

quixstreams/dataframe/windows/session.py

@@ -75,7 +75,7 @@ def process_window(
                 value=value,
                 key=key,
                 start=timestamp_ms,
-                end=timestamp_ms + timeout_ms,
+                end=timestamp_ms,  # End time is the timestamp of the last event
                 timestamp_ms=timestamp_ms,
                 late_by_ms=late_by_ms,
             )
@@ -94,13 +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
-            session_last_activity = window_end - timeout_ms
+            # window_end is stored as last_event_timestamp + 1, so subtract 1 to get actual last event time
+            session_last_activity = window_end - 1
             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 + timeout_ms
+                session_end = timestamp_ms + 1
                 can_extend_session = True
                 existing_aggregated = aggregated_value
                 old_window_to_delete = (window_start, window_end)
@@ -109,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 + timeout_ms
+            session_end = timestamp_ms + 1
 
         # Process the event for this session
         updated_windows: list[WindowKeyResult] = []
@@ -137,7 +138,7 @@ def process_window(
             updated_windows.append(
                 (
                     key,
-                    self._results(aggregated, [], session_start, session_end),
+                    self._results(aggregated, [], session_start, session_end - 1),
                 )
             )
 
@@ -219,8 +220,10 @@ def expire_by_key(
 
         windows_to_delete = []
         for (window_start, window_end), aggregated, _ in all_windows:
-            # Session expires when the session end time has passed the expiry threshold
-            if window_end <= expiry_threshold:
+            # 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:
                 collected = []
                 if collect:
                     collected = state.get_from_collection(window_start, window_end)
@@ -229,7 +232,7 @@ def expire_by_key(
                 count += 1
                 yield (
                     key,
-                    self._results(aggregated, collected, window_start, window_end),
+                    self._results(aggregated, collected, window_start, window_end - 1),
                 )
 
         # Clean up expired windows

tests/test_quixstreams/test_dataframe/test_windows/test_session.py

@@ -91,7 +91,7 @@ def test_multiaggregation(
                     key,
                     {
                         "start": 1000,
-                        "end": 11000,  # 1000 + 10000 timeout
+                        "end": 1000,  # timestamp of last event
                         "count": 1,
                         "sum": 1,
                         "mean": 1.0,
@@ -112,7 +112,7 @@ def test_multiaggregation(
                     key,
                     {
                         "start": 1000,
-                        "end": 15000,  # 5000 + 10000 timeout
+                        "end": 5000,  # timestamp of last event
                         "count": 2,
                         "sum": 5,
                         "mean": 2.5,
@@ -132,7 +132,7 @@ def test_multiaggregation(
                     key,
                     {
                         "start": 1000,
-                        "end": 15000,
+                        "end": 5000,  # timestamp of last event
                         "count": 2,
                         "sum": 5,
                         "mean": 2.5,
@@ -147,7 +147,7 @@ def test_multiaggregation(
                     key,
                     {
                         "start": 26000,
-                        "end": 36000,  # 26000 + 10000 timeout
+                        "end": 26000,  # timestamp of last event
                         "count": 1,
                         "sum": 2,
                         "mean": 2.0,
@@ -182,7 +182,7 @@ def test_sessionwindow_count(
         assert len(updated) == 1
         assert updated[0][1]["value"] == 2
         assert updated[0][1]["start"] == 1000
-        assert updated[0][1]["end"] == 15000  # 5000 + 10000
+        assert updated[0][1]["end"] == 5000  # timestamp of last event
         assert not expired
 
     @pytest.mark.parametrize("expiration", ("key", "partition"))
@@ -207,7 +207,7 @@ def test_sessionwindow_sum(
         assert len(updated) == 1
         assert updated[0][1]["value"] == 5
         assert updated[0][1]["start"] == 1000
-        assert updated[0][1]["end"] == 15000
+        assert updated[0][1]["end"] == 5000  # timestamp of last event
         assert not expired
 
     @pytest.mark.parametrize("expiration", ("key", "partition"))
@@ -232,7 +232,7 @@ def test_sessionwindow_mean(
         assert len(updated) == 1
         assert updated[0][1]["value"] == 3.0
         assert updated[0][1]["start"] == 1000
-        assert updated[0][1]["end"] == 15000
+        assert updated[0][1]["end"] == 5000  # timestamp of last event
         assert not expired
 
     @pytest.mark.parametrize("expiration", ("key", "partition"))
@@ -260,7 +260,7 @@ def test_sessionwindow_reduce(
         assert len(updated) == 1
         assert updated[0][1]["value"] == [2, 3]
         assert updated[0][1]["start"] == 1000
-        assert updated[0][1]["end"] == 15000
+        assert updated[0][1]["end"] == 5000  # timestamp of last event
         assert not expired
 
     @pytest.mark.parametrize("expiration", ("key", "partition"))
@@ -285,7 +285,7 @@ def test_sessionwindow_max(
         assert len(updated) == 1
         assert updated[0][1]["value"] == 5
         assert updated[0][1]["start"] == 1000
-        assert updated[0][1]["end"] == 15000
+        assert updated[0][1]["end"] == 5000  # timestamp of last event
         assert not expired
 
     @pytest.mark.parametrize("expiration", ("key", "partition"))
@@ -310,7 +310,7 @@ def test_sessionwindow_min(
         assert len(updated) == 1
         assert updated[0][1]["value"] == 2
         assert updated[0][1]["start"] == 1000
-        assert updated[0][1]["end"] == 15000
+        assert updated[0][1]["end"] == 5000  # timestamp of last event
         assert not expired
 
     @pytest.mark.parametrize("expiration", ("key", "partition"))
@@ -336,7 +336,7 @@ def test_sessionwindow_collect(
                 window, value=4, key=key, transaction=tx, timestamp_ms=25000
             )
         assert not updated
-        assert expired == [(key, {"start": 1000, "end": 18000, "value": [1, 2, 3]})]
+        assert expired == [(key, {"start": 1000, "end": 8000, "value": [1, 2, 3]})]
 
     @pytest.mark.parametrize(
         "timeout, grace, name",
@@ -391,7 +391,7 @@ def test_session_window_process_timeout_behavior(
             assert len(updated) == 1
             assert updated[0][1]["value"] == 1
             assert updated[0][1]["start"] == 1000
-            assert updated[0][1]["end"] == 6000  # 1000 + 5000
+            assert updated[0][1]["end"] == 1000  # timestamp of last event
             assert not expired
 
             # Add to session 1 (within timeout)
@@ -401,7 +401,7 @@ def test_session_window_process_timeout_behavior(
             assert len(updated) == 1
             assert updated[0][1]["value"] == 3
             assert updated[0][1]["start"] == 1000
-            assert updated[0][1]["end"] == 9000  # 4000 + 5000
+            assert updated[0][1]["end"] == 4000  # timestamp of last event
             assert not expired
 
             # Start session 2 (outside timeout) - should expire session 1
@@ -411,12 +411,12 @@ def test_session_window_process_timeout_behavior(
             assert len(updated) == 1
             assert updated[0][1]["value"] == 5
             assert updated[0][1]["start"] == 15000
-            assert updated[0][1]["end"] == 20000  # 15000 + 5000
+            assert updated[0][1]["end"] == 15000  # timestamp of last event
 
             assert len(expired) == 1
             assert expired[0][1]["value"] == 3
             assert expired[0][1]["start"] == 1000
-            assert expired[0][1]["end"] == 9000
+            assert expired[0][1]["end"] == 4000  # timestamp of last event
 
     def test_session_window_grace_period(
         self, session_window_definition_factory, state_manager
@@ -670,7 +670,7 @@ def test_session_window_merge_sessions(
             )
             assert len(updated) == 1
             assert updated[0][1]["start"] == 1000
-            assert updated[0][1]["end"] == 11000  # 1000 + 10000
+            assert updated[0][1]["end"] == 1000  # timestamp of last event
             assert updated[0][1]["value"] == 1
             assert not expired
 
@@ -683,12 +683,12 @@ def test_session_window_merge_sessions(
             # First session should now be expired
             assert len(expired) == 1
             assert expired[0][1]["start"] == 1000
-            assert expired[0][1]["end"] == 11000
+            assert expired[0][1]["end"] == 1000  # timestamp of last event
             assert expired[0][1]["value"] == 1
 
             assert len(updated) == 1
             assert updated[0][1]["start"] == 20000
-            assert updated[0][1]["end"] == 30000  # 20000 + 10000
+            assert updated[0][1]["end"] == 20000  # timestamp of last event
             assert updated[0][1]["value"] == 10
 
             # Add another event to the second session
@@ -697,7 +697,7 @@ def test_session_window_merge_sessions(
             )
             assert len(updated) == 1
             assert updated[0][1]["start"] == 20000
-            assert updated[0][1]["end"] == 35000  # 25000 + 10000
+            assert updated[0][1]["end"] == 25000  # timestamp of last event
             assert updated[0][1]["value"] == 15  # 10 + 5
             assert not expired
 
@@ -710,13 +710,13 @@ def test_session_window_merge_sessions(
             # Second session should be expired
             assert len(expired) == 1
             assert expired[0][1]["start"] == 20000
-            assert expired[0][1]["end"] == 35000
+            assert expired[0][1]["end"] == 25000  # timestamp of last event
             assert expired[0][1]["value"] == 15
 
             # Third session starts
             assert len(updated) == 1
             assert updated[0][1]["start"] == 50000
-            assert updated[0][1]["end"] == 60000  # 50000 + 10000
+            assert updated[0][1]["end"] == 50000  # timestamp of last event
             assert updated[0][1]["value"] == 100
 
     def test_session_window_bridging_event_scenario(
@@ -755,7 +755,7 @@ def test_session_window_bridging_event_scenario(
             )
             assert len(updated) == 1
             assert updated[0][1]["start"] == 1000
-            assert updated[0][1]["end"] == 11000  # 1000 + 10000
+            assert updated[0][1]["end"] == 1000  # timestamp of last event
             assert updated[0][1]["value"] == 5
             assert not expired
 
@@ -770,7 +770,7 @@ def test_session_window_bridging_event_scenario(
             # Event at 12000 is before 13000, so it should extend Session A
             assert len(updated) == 1
             assert updated[0][1]["start"] == 1000  # Session A extended
-            assert updated[0][1]["end"] == 22000  # 12000 + 10000
+            assert updated[0][1]["end"] == 12000  # timestamp of last event
             assert updated[0][1]["value"] == 15  # 5 + 10
             assert not expired
 
@@ -781,7 +781,7 @@ def test_session_window_bridging_event_scenario(
             # This should extend the already extended Session A further
             assert len(updated) == 1
             assert updated[0][1]["start"] == 1000  # Still Session A
-            assert updated[0][1]["end"] == 25000  # 15000 + 10000
+            assert updated[0][1]["end"] == 15000  # timestamp of last event
             assert updated[0][1]["value"] == 35  # 5 + 10 + 20
             assert not expired
 
@@ -791,7 +791,7 @@ def test_session_window_bridging_event_scenario(
             )
             assert len(expired) == 1
             assert expired[0][1]["start"] == 1000
-            assert expired[0][1]["end"] == 25000
+            assert expired[0][1]["end"] == 15000  # timestamp of last event
             assert expired[0][1]["value"] == 35  # All events combined
 
             assert len(updated) == 1
@@ -828,7 +828,7 @@ def test_session_window_string_key_extension(
             )
             assert len(updated) == 1
             assert updated[0][1]["start"] == 1000
-            assert updated[0][1]["end"] == 11000  # 1000 + 10000
+            assert updated[0][1]["end"] == 1000  # timestamp of last event
             assert updated[0][1]["value"] == 100
             assert not expired
 
@@ -839,7 +839,7 @@ def test_session_window_string_key_extension(
             )
             assert len(updated) == 1
             assert updated[0][1]["start"] == 1000  # Session extended, same start
-            assert updated[0][1]["end"] == 15000  # 5000 + 10000 (new end time)
+            assert updated[0][1]["end"] == 5000  # timestamp of last event
             assert updated[0][1]["value"] == 300  # 100 + 200
             assert not expired
 
@@ -849,7 +849,7 @@ def test_session_window_string_key_extension(
             )
             assert len(updated) == 1
             assert updated[0][1]["start"] == 1000  # Session extended again
-            assert updated[0][1]["end"] == 18000  # 8000 + 10000
+            assert updated[0][1]["end"] == 8000  # timestamp of last event
             assert updated[0][1]["value"] == 350  # 100 + 200 + 50
             assert not expired
 
@@ -861,7 +861,7 @@ def test_session_window_string_key_extension(
             assert len(updated) == 1
             assert updated[0][0] == key2  # Different key
             assert updated[0][1]["start"] == 9000
-            assert updated[0][1]["end"] == 19000  # 9000 + 10000
+            assert updated[0][1]["end"] == 9000  # timestamp of last event
             assert updated[0][1]["value"] == 75
             assert not expired
 
@@ -874,12 +874,12 @@ def test_session_window_string_key_extension(
             assert len(expired) == 1
             assert expired[0][0] == key
             assert expired[0][1]["start"] == 1000
-            assert expired[0][1]["end"] == 18000
+            assert expired[0][1]["end"] == 8000  # timestamp of last event
             assert expired[0][1]["value"] == 350
 
             # Should have started a new session for the first key
             assert len(updated) == 1
             assert updated[0][0] == key
             assert updated[0][1]["start"] == 30000
-            assert updated[0][1]["end"] == 40000
+            assert updated[0][1]["end"] == 30000  # timestamp of last event
             assert updated[0][1]["value"] == 25