BUG: Fix aligning a DataFrame with a Series with MultiIndex

When aligning a DataFrame to a Series we are using Series.reindex() to
broadcast the Series data to the new index.  That introduces NaNs when the new
index rows are not identical to the existing ones, which is not the case when
we introduce a new MultiIndex level.

In this patch we use the same approach as for aligning a Series to another
Series. That means that we have to replicate a part of Series._reindex_indexer
as DataFrame does not have it.

Author: johannes-mueller

doc/source/whatsnew/v1.5.0.rst

@@ -432,6 +432,7 @@ Reshaping
 ^^^^^^^^^
 - Bug in :func:`concat` between a :class:`Series` with integer dtype and another with :class:`CategoricalDtype` with integer categories and containing ``NaN`` values casting to object dtype instead of ``float64`` (:issue:`45359`)
 - Bug in :func:`get_dummies` that selected object and categorical dtypes but not string (:issue:`44965`)
+- Bug in :meth:`DataFrame.align` when aligning a :class:`MultiIndex` to a :class:`Series` with another :class:`MultiIndex` (:issue:`46001`)
 -
 
 Sparse

pandas/core/generic.py

@@ -114,6 +114,7 @@
 )
 
 from pandas.core import (
+    algorithms,
     arraylike,
     indexing,
     missing,
@@ -8975,10 +8976,14 @@ def _align_series(
 
         is_series = isinstance(self, ABCSeries)
 
+        if (not is_series and axis is None) or axis not in [None, 0, 1]:
+            raise ValueError("Must specify axis=0 or 1")
+
+        if is_series and axis == 1:
+            raise ValueError("cannot align series to a series other than axis 0")
+
         # series/series compat, other must always be a Series
-        if is_series:
-            if axis:
-                raise ValueError("cannot align series to a series other than axis 0")
+        if not axis:
 
             # equal
             if self.index.equals(other.index):
@@ -8988,26 +8993,40 @@ def _align_series(
                     other.index, how=join, level=level, return_indexers=True
                 )
 
-            left = self._reindex_indexer(join_index, lidx, copy)
+            if is_series:
+                left = self._reindex_indexer(join_index, lidx, copy)
+            elif join_index is None:
+                left = self.copy() if copy else self
+            else:
+                data = {
+                    c: algorithms.take_nd(
+                        self[c]._values,
+                        lidx,
+                        allow_fill=True,
+                        fill_value=None,
+                    )
+                    for c in self.columns
+                }
+                left = self._constructor(
+                    data=data, columns=self.columns, index=join_index
+                )
+
             right = other._reindex_indexer(join_index, ridx, copy)
 
         else:
+
             # one has > 1 ndim
             fdata = self._mgr
-            if axis in [0, 1]:
-                join_index = self.axes[axis]
-                lidx, ridx = None, None
-                if not join_index.equals(other.index):
-                    join_index, lidx, ridx = join_index.join(
-                        other.index, how=join, level=level, return_indexers=True
-                    )
-
-                if lidx is not None:
-                    bm_axis = self._get_block_manager_axis(axis)
-                    fdata = fdata.reindex_indexer(join_index, lidx, axis=bm_axis)
+            join_index = self.axes[1]
+            lidx, ridx = None, None
+            if not join_index.equals(other.index):
+                join_index, lidx, ridx = join_index.join(
+                    other.index, how=join, level=level, return_indexers=True
+                )
 
-            else:
-                raise ValueError("Must specify axis=0 or 1")
+            if lidx is not None:
+                bm_axis = self._get_block_manager_axis(1)
+                fdata = fdata.reindex_indexer(join_index, lidx, axis=bm_axis)
 
             if copy and fdata is self._mgr:
                 fdata = fdata.copy()

pandas/tests/frame/methods/test_align.py

@@ -243,6 +243,29 @@ def test_align_series_combinations(self):
         tm.assert_series_equal(res1, exp2)
         tm.assert_frame_equal(res2, exp1)
 
+    def test_multiindex_align_to_series_with_common_index_level(self):
+        #  GH-46001
+        foo_index = Index([1, 2, 3], name="foo")
+        bar_index = Index([1, 2], name="bar")
+
+        series = Series([1, 2], index=bar_index, name="foo_series")
+        df = DataFrame(
+            {"col": np.arange(6)},
+            index=pd.MultiIndex.from_product([foo_index, bar_index]),
+        )
+
+        expected = Series([1, 2] * 3, index=df.index, name="foo_series")
+        _, result = df.align(series, axis=0)
+
+        tm.assert_series_equal(result, expected)
+
+    def test_missing_axis_specification_exception(self):
+        df = DataFrame(np.arange(50).reshape((10, 5)))
+        s = Series(np.arange(5))
+
+        with pytest.raises(ValueError, match=r"axis=0 or 1"):
+            df.align(s)
+
     def _check_align(self, a, b, axis, fill_axis, how, method, limit=None):
         aa, ab = a.align(
             b, axis=axis, join=how, method=method, limit=limit, fill_axis=fill_axis

pandas/tests/frame/test_arithmetic.py

@@ -1072,6 +1072,120 @@ def test_frame_single_columns_object_sum_axis_1():
     tm.assert_series_equal(result, expected)
 
 
+def test_frame_multi_index_operations():
+    # GH 43321
+    df = DataFrame(
+        {2010: [1, 2, 3], 2020: [3, 4, 5]},
+        index=MultiIndex.from_product(
+            [["a"], ["b"], [0, 1, 2]], names=["scen", "mod", "id"]
+        ),
+    )
+
+    s = Series(
+        [0.4],
+        index=MultiIndex.from_product([["b"], ["a"]], names=["mod", "scen"]),
+    )
+
+    expected = DataFrame(
+        {2010: [1.4, 2.4, 3.4], 2020: [3.4, 4.4, 5.4]},
+        index=MultiIndex.from_product(
+            [["a"], ["b"], [0, 1, 2]], names=["scen", "mod", "id"]
+        ),
+    )
+    result = df.add(s, axis=0)
+
+    tm.assert_frame_equal(result, expected)
+
+
+def test_frame_multi_index_operations_series_index_to_frame_index():
+    # GH 43321
+    df = DataFrame(
+        {2010: [1], 2020: [3]},
+        index=MultiIndex.from_product([["a"], ["b"]], names=["scen", "mod"]),
+    )
+
+    s = Series(
+        [10.0, 20.0, 30.0],
+        index=MultiIndex.from_product(
+            [["a"], ["b"], [0, 1, 2]], names=["scen", "mod", "id"]
+        ),
+    )
+
+    expected = DataFrame(
+        {2010: [11.0, 21, 31.0], 2020: [13.0, 23.0, 33.0]},
+        index=MultiIndex.from_product(
+            [["a"], ["b"], [0, 1, 2]], names=["scen", "mod", "id"]
+        ),
+    )
+    result = df.add(s, axis=0)
+
+    tm.assert_frame_equal(result, expected)
+
+
+def test_frame_multi_index_operations_no_align():
+    df = DataFrame(
+        {2010: [1, 2, 3], 2020: [3, 4, 5]},
+        index=MultiIndex.from_product(
+            [["a"], ["b"], [0, 1, 2]], names=["scen", "mod", "id"]
+        ),
+    )
+
+    s = Series(
+        [0.4],
+        index=MultiIndex.from_product([["c"], ["a"]], names=["mod", "scen"]),
+    )
+
+    expected = DataFrame(
+        {2010: np.nan, 2020: np.nan},
+        index=MultiIndex.from_tuples(
+            [
+                ("a", "b", 0),
+                ("a", "b", 1),
+                ("a", "b", 2),
+                ("a", "c", np.nan),
+            ],
+            names=["scen", "mod", "id"],
+        ),
+    )
+    result = df.add(s, axis=0)
+
+    tm.assert_frame_equal(result, expected)
+
+
+def test_frame_multi_index_operations_part_align():
+    df = DataFrame(
+        {2010: [1, 2, 3], 2020: [3, 4, 5]},
+        index=MultiIndex.from_tuples(
+            [
+                ("a", "b", 0),
+                ("a", "b", 1),
+                ("a", "c", 2),
+            ],
+            names=["scen", "mod", "id"],
+        ),
+    )
+
+    s = Series(
+        [0.4],
+        index=MultiIndex.from_product([["b"], ["a"]], names=["mod", "scen"]),
+    )
+
+    expected = DataFrame(
+        {2010: [1.4, 2.4, np.nan], 2020: [3.4, 4.4, np.nan]},
+        index=MultiIndex.from_tuples(
+            [
+                ("a", "b", 0),
+                ("a", "b", 1),
+                ("a", "c", 2),
+            ],
+            names=["scen", "mod", "id"],
+        ),
+    )
+    result = df.add(s, axis=0)
+
+    tm.assert_frame_equal(result, expected)
+
+
 # -------------------------------------------------------------------
 # Unsorted
 #  These arithmetic tests were previously in other files, eventually