Fix tumbling/hopping windows with partition closing strategy

Author: gwaramadze

quixstreams/state/rocksdb/windowed/transaction.py

@@ -9,11 +9,13 @@
     PartitionTransactionStatus,
     validate_transaction_status,
 )
+from quixstreams.state.exceptions import StateSerializationError
 from quixstreams.state.metadata import DEFAULT_PREFIX, SEPARATOR
 from quixstreams.state.recovery import ChangelogProducer
 from quixstreams.state.serialization import (
     DumpsFunc,
     LoadsFunc,
+    deserialize,
     serialize,
 )
 from quixstreams.state.types import ExpiredWindowDetail, WindowDetail
@@ -360,7 +362,8 @@ def expire_all_windows(
                             end=end,
                             prefix=prefix,
                         )
-                    yield (start, end), aggregated, collected, prefix
+                    deserialized_prefix = self._deserialize_prefix(prefix)
+                    yield (start, end), aggregated, collected, deserialized_prefix
 
         else:
             # If we don't have a saved last_expired value it means one of two cases
@@ -382,7 +385,8 @@ def expire_all_windows(
                             prefix=prefix,
                         )
 
-                    yield (start, end), aggregated, collected, prefix
+                    deserialized_prefix = self._deserialize_prefix(prefix)
+                    yield (start, end), aggregated, collected, deserialized_prefix
 
         if delete:
             for prefix, start, end in to_delete:
@@ -394,6 +398,12 @@ def expire_all_windows(
             prefix=b"", cache=self._last_expired_timestamps, timestamp_ms=last_expired
         )
 
+    def _deserialize_prefix(self, prefix: bytes) -> Any:
+        try:
+            return deserialize(prefix, loads=self._loads)
+        except StateSerializationError:
+            return prefix
+
     def delete_windows(
         self, max_start_time: int, delete_values: bool, prefix: bytes
     ) -> None:

quixstreams/state/types.py

@@ -12,7 +12,7 @@
     tuple[int, int], V, bytes
 ]  # (start, end), aggregated, key
 ExpiredWindowDetail: TypeAlias = tuple[
-    tuple[int, int], V, list[V], bytes
+    tuple[int, int], V, list[V], Any
 ]  # (start, end), aggregated, collected, key