correct

Author: gwaramadze

quixstreams/state/rocksdb/timestamped.py

@@ -144,6 +144,8 @@ def set_for_timestamp(
         to the parent `set` method. The parent method internally serializes these
         into a combined key before storing the value in the update cache.
 
+        Additionally, it triggers the expiration logic.
+
         :param timestamp: Timestamp associated with the value in milliseconds.
         :param value: The value to store.
         :param prefix: The key prefix.

quixstreams/state/rocksdb/windowed/metadata.py

@@ -7,9 +7,6 @@
 LATEST_DELETED_VALUE_CF_NAME = "__value-deletion-index__"
 LATEST_DELETED_VALUE_TIMESTAMP_KEY = b"__value_deleted_start_gt__"
 
-LATEST_TIMESTAMPS_CF_NAME = "__latest-timestamps__"
-LATEST_TIMESTAMP_KEY = b"__latest_timestamp__"
-
 GLOBAL_COUNTER_CF_NAME = "__global-counter__"
 GLOBAL_COUNTER_KEY = b"__global_counter__"
 

quixstreams/state/rocksdb/windowed/partition.py

@@ -3,14 +3,14 @@
 
 from quixstreams.state.recovery import ChangelogProducer
 
+from ..metadata import LATEST_TIMESTAMPS_CF_NAME
 from ..partition import RocksDBStorePartition
 from ..types import RocksDBOptionsType
 from .metadata import (
     GLOBAL_COUNTER_CF_NAME,
     LATEST_DELETED_VALUE_CF_NAME,
     LATEST_DELETED_WINDOW_CF_NAME,
     LATEST_EXPIRED_WINDOW_CF_NAME,
-    LATEST_TIMESTAMPS_CF_NAME,
     VALUES_CF_NAME,
 )
 from .transaction import WindowedRocksDBPartitionTransaction

quixstreams/state/rocksdb/windowed/transaction.py

@@ -9,6 +9,10 @@
 from quixstreams.state.metadata import DEFAULT_PREFIX, SEPARATOR
 from quixstreams.state.recovery import ChangelogProducer
 from quixstreams.state.rocksdb.cache import CounterCache, TimestampsCache
+from quixstreams.state.rocksdb.metadata import (
+    LATEST_TIMESTAMP_KEY,
+    LATEST_TIMESTAMPS_CF_NAME,
+)
 from quixstreams.state.serialization import (
     DumpsFunc,
     LoadsFunc,
@@ -25,8 +29,6 @@
     LATEST_DELETED_WINDOW_TIMESTAMP_KEY,
     LATEST_EXPIRED_WINDOW_CF_NAME,
     LATEST_EXPIRED_WINDOW_TIMESTAMP_KEY,
-    LATEST_TIMESTAMP_KEY,
-    LATEST_TIMESTAMPS_CF_NAME,
     VALUES_CF_NAME,
 )
 from .serialization import (