Introduce delete_window method to eliminate private member usage

Author: gwaramadze

quixstreams/dataframe/windows/session.py

@@ -119,7 +119,7 @@ def process_window(
         # Delete the old window if extending an existing session
         if can_extend_session and old_window_to_delete:
             old_start, old_end = old_window_to_delete
-            transaction.delete_window(old_start, old_end, prefix=state._prefix)  # type: ignore # noqa: SLF001
+            state.delete_window(old_start, old_end)
 
         # Add to collection if needed
         if collect:
@@ -238,11 +238,7 @@ def expire_by_key(
 
         # Clean up expired windows
         for window_start, window_end in windows_to_delete:
-            state._transaction.delete_window(  # type: ignore # noqa: SLF001
-                window_start,
-                window_end,
-                prefix=state._prefix,  # type: ignore # noqa: SLF001
-            )
+            state.delete_window(window_start, window_end)
             if collect:
                 state.delete_from_collection(window_end, start=window_start)
 

quixstreams/state/rocksdb/windowed/state.py

@@ -182,3 +182,18 @@ def delete_windows(self, max_start_time: int, delete_values: bool) -> None:
             delete_values=delete_values,
             prefix=self._prefix,
         )
+
+    def delete_window(self, start_ms: int, end_ms: int) -> None:
+        """
+        Delete a specific window from the state store.
+
+        This method removes a single window entry with the specified start and end timestamps.
+
+        :param start_ms: The start timestamp of the window to delete
+        :param end_ms: The end timestamp of the window to delete
+        """
+        return self._transaction.delete_window(
+            start_ms=start_ms,
+            end_ms=end_ms,
+            prefix=self._prefix,
+        )

quixstreams/state/types.py

@@ -187,6 +187,17 @@ def delete_windows(self, max_start_time: int, delete_values: bool) -> None:
         """
         ...
 
+    def delete_window(self, start_ms: int, end_ms: int) -> None:
+        """
+        Delete a specific window from the state store.
+
+        This method removes a single window entry with the specified start and end timestamps.
+
+        :param start_ms: The start timestamp of the window to delete
+        :param end_ms: The end timestamp of the window to delete
+        """
+        ...
+
     def get_windows(
         self, start_from_ms: int, start_to_ms: int, backwards: bool = False
     ) -> list[WindowDetail[V]]:

tests/test_quixstreams/test_state/test_rocksdb/test_windowed/test_state.py

@@ -388,6 +388,26 @@ def test_delete_windows_with_values(transaction_state, get_value):
         assert get_value(timestamp_ms=2, counter=1) == "b"
 
 
+def test_delete_window(transaction_state):
+    with transaction_state() as state:
+        state.update_window(start_ms=1, end_ms=2, value=1, timestamp_ms=1)
+        state.update_window(start_ms=2, end_ms=3, value=2, timestamp_ms=2)
+        state.update_window(start_ms=3, end_ms=4, value=3, timestamp_ms=3)
+
+    with transaction_state() as state:
+        assert state.get_window(start_ms=1, end_ms=2)
+        assert state.get_window(start_ms=2, end_ms=3)
+        assert state.get_window(start_ms=3, end_ms=4)
+
+        # Delete a specific window
+        state.delete_window(start_ms=2, end_ms=3)
+
+        # Only the specified window should be deleted
+        assert state.get_window(start_ms=1, end_ms=2)
+        assert not state.get_window(start_ms=2, end_ms=3)
+        assert state.get_window(start_ms=3, end_ms=4)
+
+
 @pytest.mark.parametrize("value", [1, "string", None, ["list"], {"dict": "dict"}])
 def test_add_to_collection(transaction_state, get_value, value):
     with transaction_state() as state: