Fix: deserialization propagation after sliding window (#875)

* Fix type hinting in expire_by_key

* Fix sliding windows

Author: gwaramadze

quixstreams/dataframe/windows/sliding.py

@@ -279,7 +279,7 @@ def process_window(
         # build a complete list otherwise expired windows could be deleted
         # in state.delete_windows() and never be fetched.
         expired_windows = list(
-            self._expired_windows(state, max_expired_window_start, collect)
+            self._expired_windows(key, state, max_expired_window_start, collect)
         )
 
         state.delete_windows(
@@ -289,14 +289,14 @@ def process_window(
 
         return reversed(updated_windows), expired_windows
 
-    def _expired_windows(self, state, max_expired_window_start, collect):
+    def _expired_windows(self, key, state, max_expired_window_start, collect):
         for window in state.expire_windows(
             max_start_time=max_expired_window_start,
             delete=False,
             collect=collect,
             end_inclusive=True,
         ):
-            (start, end), (max_timestamp, aggregated), collected, key = window
+            (start, end), (max_timestamp, aggregated), collected, _ = window
             if end == max_timestamp:
                 yield key, self._results(aggregated, collected, start, end)
 

quixstreams/dataframe/windows/time_based.py

@@ -229,7 +229,7 @@ def expire_by_partition(
 
     def expire_by_key(
         self,
-        key: bytes,
+        key: Any,
         state: WindowedState,
         max_expired_start: int,
         collect: bool,