concat(): Require timestamp alignment only when SDFs have different topics (#866)

Author: daniil-quix

quixstreams/dataframe/dataframe.py

@@ -2,6 +2,7 @@
 
 import contextvars
 import functools
+import itertools
 import operator
 import pprint
 import typing
@@ -1644,7 +1645,13 @@ def concat(self, other: "StreamingDataFrame") -> "StreamingDataFrame":
         """
 
         merged_stream = self.stream.merge(other.stream)
-        self._registry.require_time_alignment()
+
+        # Enable partition time alignment only when concatenated StreamingDataFrames
+        # have different topics (they could be just branches)
+        total_topics = {t.name for t in itertools.chain(self.topics, other.topics)}
+        if len(total_topics) > 1:
+            self._registry.require_time_alignment()
+
         return self.__dataframe_clone__(
             *self.topics, *other.topics, stream=merged_stream
         )

tests/test_quixstreams/test_dataframe/test_dataframe.py

@@ -2558,7 +2558,8 @@ def test_concat_same_topic_success(self, topic_manager_factory, dataframe_factor
         topic_manager = topic_manager_factory()
         topic = topic_manager.topic(str(uuid.uuid4()))
 
-        sdf = dataframe_factory(topic)
+        registry = DataFrameRegistry()
+        sdf = dataframe_factory(topic, registry=registry)
         sdf_branch1 = sdf.apply(lambda v: v + 1)
         sdf_branch2 = sdf.apply(lambda v: v + 2)
 
@@ -2570,6 +2571,8 @@ def test_concat_same_topic_success(self, topic_manager_factory, dataframe_factor
             (2, b"key1", 1, None),
             (3, b"key1", 1, None),
         ]
+        # Timestamp alignment is not required the concated SDFs are branches
+        assert not registry.requires_time_alignment
 
     def test_concat_stateful_success(
         self,