Fix concat axis 1 bug in divisions

dask_expr/_concat.py

@@ -136,7 +136,9 @@ def _lower(self):
         dfs = self._frames
         if self.axis == 1:
             if self._are_co_alinged_or_single_partition:
-                return ConcatIndexed(self.ignore_order, self._kwargs, self.axis, *dfs)
+                return ConcatIndexed(
+                    self.ignore_order, self._kwargs, self.axis, self.join, *dfs
+                )
 
             elif (
                 all(not df.known_divisions for df in dfs)
@@ -149,7 +151,9 @@ def _lower(self):
                         " are \n aligned. This assumption is not generally "
                         "safe."
                     )
-                return ConcatUnindexed(self.ignore_order, self._kwargs, self.axis, *dfs)
+                return ConcatUnindexed(
+                    self.ignore_order, self._kwargs, self.axis, self.join, *dfs
+                )
             elif self._all_known_divisions:
                 from dask_expr._repartition import Repartition
 
@@ -338,25 +342,29 @@ def _layer(self):
 
 
 class ConcatUnindexed(Blockwise):
-    _parameters = ["ignore_order", "_kwargs", "axis"]
-    _defaults = {"ignore_order": False, "_kwargs": {}, "axis": 1}
-    _keyword_only = ["ignore_order", "_kwargs", "axis"]
+    _parameters = ["ignore_order", "_kwargs", "axis", "join"]
+    _defaults = {"ignore_order": False, "_kwargs": {}, "axis": 1, "join": "outer"}
+    _keyword_only = ["ignore_order", "_kwargs", "axis", "join"]
 
     @functools.cached_property
     def _meta(self):
         return methods.concat(
             [df._meta for df in self.dependencies()],
             ignore_order=self.ignore_order,
             axis=self.axis,
+            join=self.join,
             **self.operand("_kwargs"),
         )
 
     @staticmethod
-    def operation(*args, ignore_order, _kwargs, axis):
+    def operation(*args, ignore_order, _kwargs, axis, join):
         return concat_and_check(args, ignore_order=ignore_order)
 
 
 class ConcatIndexed(ConcatUnindexed):
     @staticmethod
-    def operation(*args, ignore_order, _kwargs, axis):
-        return methods.concat(args, ignore_order=ignore_order, axis=axis)
+    def operation(*args, ignore_order, _kwargs, axis, join):
+        return methods.concat(args, ignore_order=ignore_order, axis=axis, join=join)
+
+    def _broadcast_dep(self, dep: Expr):
+        return dep.npartitions == 1

dask_expr/tests/test_concat.py

@@ -1,7 +1,7 @@
 import numpy as np
 import pytest
 
-from dask_expr import DataFrame, FrameBase, Len, Series, concat, from_pandas
+from dask_expr import DataFrame, FrameBase, Len, Series, concat, from_dict, from_pandas
 from dask_expr.tests._util import _backend_library, assert_eq
 
 # Set DataFrame backend for this module
@@ -349,3 +349,14 @@ def test_concat_series_and_projection(df, pdf):
     result = concat([df.x, df.y], axis=1)[["x"]]
     expected = pd.concat([pdf.x, pdf.y], axis=1)[["x"]]
     assert_eq(result, expected)
+
+
+@pytest.mark.parametrize("npartitions", [1, 2])
+@pytest.mark.parametrize("join", ["inner", "outer"])
+def test_concat_single_partition_known_divisions(join, npartitions):
+    df1 = from_dict({"a": [1, 2, 3], "b": [1, 2, 3]}, npartitions=npartitions)
+    df2 = from_dict({"c": [1, 2]}, npartitions=npartitions)
+
+    result = concat([df1, df2], axis=1, join=join)
+    expected = pd.concat([df1.compute(), df2.compute()], axis=1, join=join)
+    assert_eq(result, expected)