Add support for session window #917

petrpan26 · 2025-06-03T10:52:44Z

Session Windows Implementation for Quix Streams

This PR introduces Session Windows - dynamic windowing that groups events by activity sessions rather than fixed time intervals.

Key Features

Activity-driven sessions: Sessions extend automatically when events arrive within timeout period
Flexible configuration: Configurable timeout and grace periods for late events
Full aggregation support: Works with all existing aggregation functions (sum, count, mean, max, min, reduce, collect)
Both modes supported: .current() for real-time updates, .final() for complete sessions
Closing strategies: Both "key" and "partition" level session expiration

Usage Examples

# User session tracking (30min timeout, 5min grace)
sdf.session_window(timeout_ms=timedelta(minutes=30), grace_ms=timedelta(minutes=5)) \
   .agg(page_views=Count(), actions=Collect("action_type")) \
   .final()

# Real-time fraud detection
sdf.session_window(timeout_ms=timedelta(minutes=10)) \
   .agg(transaction_count=Count(), total_amount=Sum("amount")) \
   .current()

Use Cases Enabled

User activity tracking (web/app sessions)
Fraud detection (transaction patterns)
IoT device monitoring (activity sessions)
Gaming analytics (player sessions)

Testing & Documentation

✅ 32 comprehensive tests covering all session window functionality
✅ Complete documentation added to docs/windowing.md with examples
✅ Session merging scenarios and edge cases tested
✅ Multi-key session management validated

Technical Implementation

Smart session extension logic with grace period support
Efficient RocksDB-backed state management
Proper transaction handling for concurrent access
Memory-efficient session garbage collection

This enables sophisticated streaming analytics use cases that were previously difficult to implement with fixed-time windows.

gwaramadze

Hello. Once again, thanks for your contributions. Here is the first batch of comments. Please take a look. We will continue to review and test it.

gwaramadze · 2025-06-11T08:27:27Z

docs/windowing.md

@@ -9,7 +9,9 @@ With windows, you can calculate such aggregations as:
 - Total of website visitors for every hour
 - The average speed of a vehicle over the last 10 minutes
 - Maximum temperature of a sensor observed over 30 second ranges
- Give an user a reward after 10 succesful actions 
+- Give an user a reward after 10 succesful actions


Suggested change

- Give an user a reward after 10 succesful actions

- Give a user a reward after 10 successful actions

gwaramadze · 2025-06-11T08:45:48Z

docs/windowing.md

+Grace:   2 seconds
+
+Session 1: [0, 20] - Events A, B (B extends the session from A)
+Session 2: [25, 35] - Events C, D (D extends the session from C)  


shouldn't that be [25, 40]?

Suggested change

Session 2: [25, 35] - Events C, D (D extends the session from C)

Session 2: [25, 40] - Events C, D (D extends the session from C)

gwaramadze · 2025-06-11T10:28:08Z

quixstreams/dataframe/windows/definitions.py

+    @property
+    def timeout_ms(self) -> int:
+        return self._timeout_ms
+


I don't think this property is used anywhere

Suggested change

@property

def timeout_ms(self) -> int:

return self._timeout_ms

gwaramadze · 2025-06-11T11:11:07Z

quixstreams/dataframe/windows/time_based.py

+            partition = ctx.partition
+            offset = ctx.offset
+        except:
+            # In test environments, message context might not be available


We must not mock objects for test purposes conditionally in the lib code. Sliding windows have mock_message_context, so it should be reused where necessary.

gwaramadze · 2025-06-11T11:53:39Z

quixstreams/dataframe/windows/time_based.py

+                round(time.monotonic() - start, 2),
+            )
+
+    def _on_expired_session(


This method is almost identical to existing _on_expired_window, we should abstract it.

gwaramadze · 2025-06-11T12:48:17Z

quixstreams/dataframe/windows/time_based.py

+        # By this point, session_start and session_end are guaranteed to be set
+        assert session_start is not None  # noqa: S101
+        assert session_end is not None  # noqa: S101
+
+        # Output intermediate results for aggregations
+        if aggregate:


Asserts are unnecessary. If we let them go, the next if is also rendered redundant.

Suggested change

# By this point, session_start and session_end are guaranteed to be set

assert session_start is not None # noqa: S101

assert session_end is not None # noqa: S101

# Output intermediate results for aggregations

if aggregate:

gwaramadze · 2025-06-11T13:10:16Z

quixstreams/dataframe/windows/time_based.py

+
+        # Clean up expired windows
+        for window_start, window_end in windows_to_delete:
+            state._transaction.delete_window(  # type: ignore # noqa: SLF001


Usage of the private attribute _transaction, this must be fixed.

gwaramadze · 2025-06-12T08:33:28Z

quixstreams/dataframe/dataframe.py

@@ -1484,6 +1485,103 @@ def sliding_count_window(
            name=name,
        )

+    def session_window(
+        self,
+        timeout_ms: Union[int, timedelta],


A more widely adopted nomenclature for session windows is "gap". Let's rename it to gap_ms or a more explicit inactivity_gap_ms.

gwaramadze · 2025-06-12T09:43:29Z

quixstreams/dataframe/windows/aggregations.py

@@ -348,6 +348,7 @@ def __init__(
        reducer: Callable[[R, Any], R],
        initializer: Callable[[Any], R],
    ) -> None:
+        super().__init__()


Wasn't this a part of a separate bugfix PR? If not, then it should be separated together with the corresponding test case.

gwaramadze · 2025-06-13T08:35:50Z

quixstreams/dataframe/windows/time_based.py

+            state._transaction.delete_window(  # type: ignore # noqa: SLF001
+                window_start,
+                window_end,
+                prefix=state._prefix,  # type: ignore # noqa: SLF001


also here, we can't use the private member like this

Add fix for using reducer with window with test added

4d14faf

petrpan26 force-pushed the petrpan26/session-window branch from d2b4108 to a8b14a2 Compare June 3, 2025 15:38

Hoang Phan added 2 commits June 3, 2025 11:41

Fix precommit

027b132

Add support for session window

07ac02a

petrpan26 force-pushed the petrpan26/session-window branch from a8b14a2 to 07ac02a Compare June 3, 2025 15:41

Add fix for serializer

ab220fc

gwaramadze reviewed Jun 13, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Add support for session window #917

Add support for session window #917

Uh oh!

petrpan26 commented Jun 3, 2025

Uh oh!

gwaramadze left a comment

Uh oh!

gwaramadze Jun 11, 2025

Uh oh!

gwaramadze Jun 11, 2025

Uh oh!

gwaramadze Jun 11, 2025

Uh oh!

gwaramadze Jun 11, 2025

Uh oh!

gwaramadze Jun 11, 2025

Uh oh!

gwaramadze Jun 11, 2025

Uh oh!

gwaramadze Jun 11, 2025

Uh oh!

gwaramadze Jun 12, 2025

Uh oh!

gwaramadze Jun 12, 2025

Uh oh!

gwaramadze Jun 13, 2025

Uh oh!

Uh oh!

	- Give an user a reward after 10 succesful actions
	- Give a user a reward after 10 successful actions

	Session 2: [25, 35] - Events C, D (D extends the session from C)
	Session 2: [25, 40] - Events C, D (D extends the session from C)

	@property
	def timeout_ms(self) -> int:
	return self._timeout_ms

Add support for session window #917

Are you sure you want to change the base?

Add support for session window #917

Uh oh!

Conversation

petrpan26 commented Jun 3, 2025

Session Windows Implementation for Quix Streams

Key Features

Usage Examples

Use Cases Enabled

Testing & Documentation

Technical Implementation

Uh oh!

gwaramadze left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!