adjust the changelog name for named window operations to accommodate branching (#563)

Author: tim-quix

quixstreams/dataframe/windows/definitions.py

@@ -286,10 +286,8 @@ def __init__(
         )
 
     def _get_name(self, func_name: str) -> str:
-        return (
-            self._name
-            or f"hopping_window_{self._duration_ms}_{self._step_ms}_{func_name}"
-        )
+        prefix = f"{self._name}_hopping_window" if self._name else "hopping_window"
+        return f"{prefix}_{self._duration_ms}_{self._step_ms}_{func_name}"
 
     def _create_window(
         self,
@@ -321,7 +319,8 @@ def __init__(
         )
 
     def _get_name(self, func_name: str) -> str:
-        return self._name or f"tumbling_window_{self._duration_ms}_{func_name}"
+        prefix = f"{self._name}_tumbling_window" if self._name else "tumbling_window"
+        return f"{prefix}_{self._duration_ms}_{func_name}"
 
     def _create_window(
         self,

quixstreams/state/manager.py

@@ -168,7 +168,10 @@ def register_windowed_store(self, topic_name: str, store_name: str):
         """
         store = self._stores.get(topic_name, {}).get(store_name)
         if store:
-            raise WindowedStoreAlreadyRegisteredError()
+            raise WindowedStoreAlreadyRegisteredError(
+                "This window range and type combination already exists; "
+                "to use this window, provide a unique name via the `name` parameter."
+            )
         self._stores.setdefault(topic_name, {})[store_name] = WindowedRocksDBStore(
             name=store_name,
             topic=topic_name,

tests/test_quixstreams/test_app.py

@@ -1511,6 +1511,9 @@ def test_changelog_recovery_window_store(
         store_name = "window"
         window_duration_ms = 5000
         window_step_ms = 2000
+        actual_store_name = (
+            f"{store_name}_hopping_window_{window_duration_ms}_{window_step_ms}_sum"
+        )
         msg_tick_ms = 1000
         msg_int_value = 10
         partition_timestamps = {
@@ -1591,13 +1594,13 @@ def validate_state():
             with state_manager_factory(
                 group_id=consumer_group, state_dir=state_dir
             ) as state_manager:
-                state_manager.register_windowed_store(topic.name, store_name)
+                state_manager.register_windowed_store(topic.name, actual_store_name)
                 for p_num, windows in expected_window_updates.items():
                     state_manager.on_partition_assign(
                         topic=topic.name, partition=p_num, committed_offset=-1001
                     )
                     store = state_manager.get_store(
-                        topic=topic.name, store_name=store_name
+                        topic=topic.name, store_name=actual_store_name
                     )
 
                     # in this test, each expiration check only deletes one window,

tests/test_quixstreams/test_dataframe/test_windows/test_hopping.py

@@ -21,7 +21,7 @@ class TestHoppingWindow:
     @pytest.mark.parametrize(
         "duration, grace, step, provided_name, func_name, expected_name",
         [
-            (10, 5, 3, "custom_window", "sum", "custom_window"),
+            (10, 5, 3, "custom_window", "sum", "custom_window_hopping_window_10_3_sum"),
             (10, 5, 3, None, "sum", "hopping_window_10_3_sum"),
             (15, 5, 3, None, "count", "hopping_window_15_3_count"),
         ],

tests/test_quixstreams/test_dataframe/test_windows/test_tumbling.py

@@ -19,7 +19,7 @@ class TestTumblingWindow:
     @pytest.mark.parametrize(
         "duration, grace, provided_name, func_name, expected_name",
         [
-            (10, 5, "custom_window", "sum", "custom_window"),
+            (10, 5, "custom_window", "sum", "custom_window_tumbling_window_10_sum"),
             (10, 5, None, "sum", "tumbling_window_10_sum"),
             (15, 5, None, "count", "tumbling_window_15_count"),
         ],