Feature: Session windows

[gwaramadze]

Session windows

Summary

Adds StreamingDataFrame.session_window(): events of one message key are grouped
into sessions separated by more than inactivity_gap_ms, with dynamic boundaries
driven by the data rather than a fixed schedule. Sessions support the full
aggregation surface (.agg(), .sum(), .count(), .reduce(), .collect(),
…), both closing strategies (.final("key") / .final("partition")), .current()
and the on_late callback.

The first implementation on this branch fragmented long sessions, dropped valid out-of-order
events, never merged sessions bridged by a late event, and scanned the whole
partition on every expiry. process_window is rebuilt around a two-neighbour
probe, and the supporting state primitives are added alongside it.

Semantics

• An event joins a session iff it is within one inactivity gap of either boundary.
  grace_ms no longer takes part in assignment, only in closing. An event is late
  iff ts < watermark - gap - grace, so the default grace_ms=0 still leaves a
  full gap of out-of-order tolerance.
• An out-of-order event within one gap of two open sessions merges them.
• Stored sessions are maximal, disjoint and never within one gap of each other,
  including a session starting at timestamp 0.
• Session end is half-open (last event + 1), like every other window type.
• A session closes once the watermark passes last event + 2 * gap + grace — the
  latest an admissible event could still extend it.
• Trigger callbacks (before_update / after_update, PR Feature: Early Window Expiration with Triggers #1044) are rejected with
  ValueError, mirroring the existing sliding-window rejection: a bridging event
  matches up to two sessions, so there is no single "current value", and a forced
  early close would break the disjoint/non-adjacent guarantee.

State-layer changes

• iter_windows(prefix, start_from_ms, start_to_ms=None, backwards=False) —
  lazy, inclusive-lower-bound, optionally unbounded above, merged with the
  transaction's uncommitted updates. Replaces the materialising get_windows
  in the session hot path.
• iter_prefixes() — streams the stored window keys once and yields each distinct
  message-key prefix, replacing the full keys() scan in partition-mode expiry.
• get_expiry_checkpoint / set_expiry_checkpoint — persisted cursors, so
  expire_by_key is O(log n + expired) and expire_by_partition is gated to
  O(1) per message in the common case.
• RocksDBStorePartition.iter_items(upper_bound=None) — unbounded forward
  iteration. The pinned rocksdict yields nothing when only a lower bound is set
  on ReadOptions, so the two bounds are now set both-or-neither and from_key
  carries forward correctness; backwards=True without an upper bound raises.
• WindowedState.delete_window(start_ms, end_ms) — a session is re-keyed whenever
  it is extended or merged, so the old key has to go.
• BaseAggregator.merge() / mergeable, implemented on all nine built-in
  aggregators, plus Reduce(merger=). A non-mergeable aggregation raises
  InvalidOperation when the session window is defined.

Regression tests

Every row fails on the code without its fix. All live in
tests/test_quixstreams/test_dataframe/test_windows/test_session.py unless
another path is given.

Defect	Test	Failure without the fix
Long session fragmented after 2 * gap	test_long_session_does_not_fragment	assert updated[0][1]["start"] == 1000 — a new session starts mid-stream
Partition mode accepted events below the watermark	test_partition_strategy_rejects_events_before_partition_watermark	on_late never called; the event is folded into a live session
grace_ms=0 dropped in-gap out-of-order events	test_out_of_order_event_inside_gap_is_accepted_with_zero_grace	on_late.call_count == 0 fails; the event is reported late
Bridging event did not merge two sessions	test_bridging_event_merges_two_sessions, test_session_window_merge_sessions, test_bridging_event_merges_two_sessions_multiaggregation	len(stored_windows(...)) == 1 fails — two sessions remain
Session starting at timestamp 0 not found by the backwards probe	test_session_starting_at_epoch_zero_extends	a second session [1000, 1001) is stored instead of extending [0, …)
Stored sessions within one gap of each other	test_stored_sessions_are_never_within_the_gap	the non-adjacency assertion fires on the surviving state
Session end inclusive	test_session_end_is_exclusive, test_session_end_is_exclusive_collect_no_leaked_values	end == 5001 fails; the collect variant leaves one value undeleted
grace_ms had no type check	test_session_window_def_init_invalid_grace_type	no TypeError raised for grace_ms="invalid"
Non-mergeable aggregation accepted	test_non_mergeable_aggregation_rejected_at_definition_time, test_sessionwindow_reduce_without_merger_raises	no InvalidOperation at definition time; the merge corrupts at runtime
Prefix ambiguity — unbounded lookup leaked another key's windows	test_windowed/test_transaction.py::TestIterWindows::test_prefix_leak_for_separator_extended_key	`b"user
Prefix ambiguity — enumerator seeked past an extended key	test_windowed/test_transaction.py::TestIterPrefixes::test_iter_prefixes_skips_separator_extended_key	`prefixes == [b"user", b"user
Closing and lateness shared one threshold	test_close_before_shared_by_lateness_and_closing_allows_adjacent_sessions	two emitted sessions 1 ms apart trip the non-adjacency assertion
Reaggregating a stored None crashed	test_multiaggregation_reaggregation_of_none_value, test_merge_of_none_value	TypeError inside _aggregate_value / _merge_values on the persisted None
Null partition checkpoint overwritten by the first message	test_partition_checkpoint_none_defers_already_due_sessions	{b"k1", b"k2"} <= expired_keys fails; the due sweep is deferred
Partition watermark ignored per-key history	test_partition_watermark_ignores_prior_key_mode_history	a second session overlapping the already-emitted [100, 106) is accepted
Expiry cursor stranded sessions when grace_ms grew	test_cursor_not_invalidated_when_grace_regresses_across_reopen	session [95, 96) is never emitted, however far the watermark advances
Reduce.mergeable ignored an overridden merge()	test_reduce_mergeable_true_when_merge_overridden_without_merger_kwarg (+ …_builds)	mergeable is True fails; the definition raises InvalidOperation
Reduce accepted a conflicting merger= + merge() override	test_reduce_conflicting_merge_override_not_caught_at_definition_time	nothing raises at definition time; the broken override runs mid-stream
Expiry paths scanned the partition per message	test_expiry_paths_do_not_scan_the_partition	get_windows / keys spies record calls
iter_windows exclusive lower bound / materialising	test_windowed/test_transaction.py::TestIterWindows::test_inclusive_lower_bound, …::test_is_lazy_does_not_use_materializing_get_items	window at start 0 missing; _get_items called for the first element
iter_items could not iterate unbounded	test_rocksdb_partition.py::test_iter_items_unbounded_upper_bound, …::test_iter_items_backwards_requires_upper_bound	zero rows returned; backwards + unbounded mis-seeks silently
delete_window missing from WindowedState	test_windowed/test_state.py::test_delete_window	AttributeError

Docs

• docs/windowing.md: new "Session Windows" section (!!! info New in v3.27.0)
  covering the assignment and closing rules, merging and its effect on .current()
  consumers, the mergeable-aggregator list, the grace period, and both closing
  strategies. Existing sections updated for the new window type: the on_late
  list, closing strategies, store naming (session_window_30000) and the
  window-definition-update notes.
• Docstrings: StreamingDataFrame.session_window, SessionWindowDefinition,
  SessionWindow, BaseAggregator.merge / mergeable, Reduce(merger=),
  WindowDefinition.reduce, and the new WindowedState /
  WindowedPartitionTransaction protocol methods. docs/api-reference/*.md is
  left to the post-merge regeneration.

Partition-mode expiry emitted the store prefix as the message key

Found while running the feature on a real deployment, and fixed here because
session windows hit it the moment closing_strategy="partition" is used.

as_state(prefix=key) stores a non-bytes key as serialize(key, dumps), so a
str key user-1 reaches the store as b'"user-1"'. Both partition-mode sweeps
-- TimeWindow.expire_by_partition for tumbling and hopping windows, and
SessionWindow.expire_by_partition -- then yielded that prefix downstream as the
message key, while the key-scoped path yielded the real key. Switching closing
strategy therefore changed the emitted key's value and type with a correct-looking
payload, breaking any downstream group_by, join or keyed consumer silently.

WindowedPartitionTransaction.key_from_prefix() reverses as_state() and both
sweeps use it. Nothing the store writes changes, so existing state is read
unchanged. The discriminator is the processed record's key type, which is what
as_state() applies when writing: exact while a topic's key deserializer is
type-stable, which holds for every built-in, and group_by re-keys through a JSON
repartition topic. Persisting the key kind would remove the assumption at the cost
of an on-disk format change.

Tumbling and hopping windows carry the same defect on main today and are fixed
by the same commit; sliding windows reject partition mode and count windows have
no closing strategy. Regression tests for all three window types are in
tests/test_quixstreams/test_dataframe/test_windows/test_partition_expiry_key.py,
red before the fix.

Verified on a live deployment

A Quix Cloud pipeline ran a scripted event table through three deployments of one
application differing only by closing mode, with a real RocksDB state volume and a
real mid-run restart. Every emitted session was compared against expectations
derived by hand from the three rules above.

scenario	evidence
A continuous stream stays one session	10 events one gap-half apart -> count 10, one session, no fragmentation
An in-gap out-of-order event joins	seqs [0, 1, 3, 2] in a single session, event-time ordered
A bridging event merges two sessions	two stored sessions replaced by one spanning both, aggregates combined
A late event is dropped once	on_late fired once; no second session, no resurrection
An idle key closes only under partition	the same key is emitted by the partition-mode deployment and never by the key-mode one
.current() supersession after a merge	an update at start=1400000 followed by one at start=1200000 for the same key
A session survives a restart	stopped and restarted mid-session; one session, count 10, emitted exactly once

The partition-mode key defect above was found by this rig, not by the unit tests:
those compare against the same prefix representation the code produces, so the
mismatch is invisible to them. After the fix the same rig, same data and same
oracle turned the partition arm from zero matched sessions to all matched, with
identical keys on the key-mode and partition-mode output topics.

Known limitation

A partition-mode expiry sweep enumerates prefixes with a linear pass over the
stored window keys, so its cost is O(stored windows) rather than
O(prefixes). No seek can step past one prefix's key range without risking
another's, because a SEPARATOR-extended message key's window keys sort inside the
shorter key's range. The checkpoint gate amortises the pass, and closed sessions
are deleted so only open sessions remain in the column family.

test_expiry_scan_cost_grows_with_windows_per_key_not_with_key_count is marked
xfail(strict=True) and documents the bound an (end, prefix, start) index
column family would restore. That index is the follow-up, not part of this PR.

Compatibility

No syntax newer than the repository's current Python floor is introduced; the
changed modules parse under it unchanged.

[davizucon]

Hi, this is a very promising PR!
Just wanted to check in and see if there was any update on it.

[lpsinger]

I am evaluating Quix for a project that requires this feature. Is there any update on it?

[LudvikQuix]

hi guys, sorry for late response, there were some staff changes. I am working on this PR to be merged.