Self joins not supported

Author: gwaramadze

quixstreams/dataframe/dataframe.py

@@ -1661,6 +1661,12 @@ def join_latest(
         merger: Optional[Callable[[Any, Any], Any]] = None,
         retention_ms: Union[int, timedelta] = timedelta(days=7),
     ) -> "StreamingDataFrame":
+        if self.stream_id == right.stream_id:
+            raise ValueError(
+                "Joining dataframes originating from "
+                "the same topic is not yet supported.",
+            )
+
         if how not in get_args(JoinHow):
             raise ValueError(
                 f"Invalid how value: {how}. "

tests/test_quixstreams/test_dataframe/test_dataframe.py

@@ -2904,3 +2904,19 @@ def test_retention_ms(
 
         assert publish_left(timestamp=4) == []
         assert publish_left(timestamp=5) == [({"left": 4, "right": 2}, b"key", 5, None)]
+
+    def test_self_join_not_supported(self, create_topic, create_sdf):
+        topic = create_topic()
+        match = (
+            "Joining dataframes originating from the same topic is not yet supported."
+        )
+
+        # The very same sdf object
+        sdf = create_sdf(topic)
+        with pytest.raises(ValueError, match=match):
+            sdf.join_latest(sdf)
+
+        # Same topic, different branch
+        sdf2 = sdf.apply(lambda v: v)
+        with pytest.raises(ValueError, match=match):
+            sdf.join_latest(sdf2)