Deserialize key for partition closing strategy

gwaramadze · gwaramadze · commit dd3bf22a31a6 · 2025-04-18T10:42:09.000+02:00
diff --git a/quixstreams/dataframe/windows/time_based.py b/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,
diff --git a/quixstreams/state/rocksdb/windowed/transaction.py b/quixstreams/state/rocksdb/windowed/transaction.py
@@ -14,6 +14,7 @@
 from quixstreams.state.serialization import (
     DumpsFunc,
     LoadsFunc,
+    deserialize,
     serialize,
 )
 from quixstreams.state.types import ExpiredWindowDetail, WindowDetail
@@ -360,6 +361,7 @@ def expire_all_windows(
                             end=end,
                             prefix=prefix,
                         )
+                    prefix = deserialize(prefix, loads=self._loads)
                     yield (start, end), aggregated, collected, prefix
 
         else:

Original file line number	Diff line number	Diff line change
`@@ -14,6 +14,7 @@`
`14`	`14`	`from quixstreams.state.serialization import (`
`15`	`15`	`DumpsFunc,`
`16`	`16`	`LoadsFunc,`
	`17`	`+ deserialize,`
`17`	`18`	`serialize,`
`18`	`19`	`)`
`19`	`20`	`from quixstreams.state.types import ExpiredWindowDetail, WindowDetail`
`@@ -360,6 +361,7 @@ def expire_all_windows(`
`360`	`361`	`end=end,`
`361`	`362`	`prefix=prefix,`
`362`	`363`	`)`
	`364`	`+ prefix = deserialize(prefix, loads=self._loads)`
`363`	`365`	`yield (start, end), aggregated, collected, prefix`
`364`	`366`
`365`	`367`	`else:`