Repository: modin-project/modin
Pull request: FIX-#4660: Fix `fillna` when Modin series object is an argument
Signed-off-by: Myachev <anatoly.myachev@intel.com>

<!--
Thank you for your contribution!
Please review the contributing docs: https://modin.readthedocs.io/en/latest/development/contributing.html
if you have questions about contributing.
-->

## What do these changes do?

<!-- Please give a short brief about these changes. -->

- [x] commit message follows format outlined [here](https://modin.readthedocs.io/en/latest/development/contributing.html#commit-message-formatting)
- [x] passes `flake8 modin/ asv_bench/benchmarks scripts/doc_checker.py`
- [x] passes `black --check modin/ asv_bench/benchmarks scripts/doc_checker.py`
- [x] signed commit with `git commit -s` <!-- you can amend your commit with a signature via `git commit -amend -s` -->
- [x] Resolves #4660 <!-- issue must be created for each patch -->
- [x] tests added and passing
- [x] module layout described at `docs/development/architecture.rst` is up-to-date <!-- if you have added, renamed or removed files or directories please update the documentation accordingly -->
- [x] added (Issue Number: PR title (PR Number)) and github username to release notes for next major release <!-- e.g. DOCS-#4077: Add release notes template to docs folder (#4078) -->


Problem statement:
Fillna doesn't work when series object is an argument
### System information
- **OS Platform and Distribution (e.g., Linux Ubuntu 16.04)**:
- **Modin version** (`modin.__version__`):
- **Python version**:
- **Code we can use to reproduce**:

```python
import modin.pandas as pd

df = pd.DataFrame({'a': ['a'], 'b': ['b'],}, index=['row1'])
df['c'] = pd.NA
df['c'] = df['c'].fillna(df['b'])   # df['c'].fillna('c') works fine
print(df)
```

<!--
You can obtain the Modin version with

python -c "import modin; print(modin.__version__)"
-->

### Describe the problem
<!-- Describe the problem clearly here. -->

when fillna is called with series object as argument seems nothing happens while pandas does actual na replacement

### Source code / logs
<!-- Include any logs or source code that would be helpful to diagnose the problem. If including tracebacks, please include the full traceback. Large logs and files should be attached. Try to provide a reproducible test case that is the bare minimum necessary to generate the problem. -->

Hints from issue discussion:
As far as I can tell, it seems like something's going awry in `broadcast_axis_partitions`, on [this line](https://github.com/modin-project/modin/blob/35782883efd7e22f8e29a7c9887a4453adabf783/modin/core/dataframe/pandas/partitioning/partition_manager.py#L462): the value in the partition seems to be changed when the dataframe is reindexed. I'll let someone more familiar with the codebase weigh in on why this is the case.

<details>
<summary>Abridged stack trace:</summary>
<pre>
-> print(df['c'].fillna(df['b']))  # df['c'].fillna('c') works fine
  modin/modin/logging/logger_decorator.py(128)run_and_log()
-> return obj(*args, **kwargs)
  modin/modin/pandas/series.py(993)fillna()
-> return super(Series, self)._fillna(
  modin/modin/logging/logger_decorator.py(128)run_and_log()
-> return obj(*args, **kwargs)
  modin/modin/pandas/base.py(1525)_fillna()
-> new_query_compiler = self._query_compiler.fillna(
  modin/modin/logging/logger_decorator.py(128)run_and_log()
-> return obj(*args, **kwargs)
  modin/modin/core/storage_formats/pandas/query_compiler.py(1882)fillna()
-> new_modin_frame = self._modin_frame.binary_op(
  modin/modin/logging/logger_decorator.py(128)run_and_log()
-> return obj(*args, **kwargs)
  modin/modin/core/dataframe/pandas/dataframe/dataframe.py(115)run_f_on_minimally_updated_metadata()
-> result = f(self, *args, **kwargs)
  modin/modin/core/dataframe/pandas/dataframe/dataframe.py(2526)binary_op()
-> ) = new_left_frame._copartition(1, new_right_frame, join_type, sort=True)
  modin/modin/logging/logger_decorator.py(128)run_and_log()
-> return obj(*args, **kwargs)
  modin/modin/core/dataframe/pandas/dataframe/dataframe.py(2471)_copartition()
-> reindexed_other_list[i] = other_frames[
  modin/modin/core/dataframe/pandas/partitioning/partition_manager.py(573)map_axis_partitions()
-> return cls.broadcast_axis_partitions(
> modin/modin/core/dataframe/pandas/partitioning/partition_manager.py(476)broadcast_axis_partitions()
</pre>
</details>
After [this commit](https://github.com/modin-project/modin/commit/8f35ab57996d18f2c1f001d624682952853454cd), `_copartition` function is called in `binary_op` also for columns, which still does reindexing, however, for some functions this action is superfluous, for example, in the case of `fillna`.

Should we make reindexing optional or make a second `_copartition` call optional? cc @prutskov 
@anmyachev, according to our patterns of parallel execution partitioning of operands must be aligned by both axis for binary operator (https://modin.readthedocs.io/en/latest/flow/modin/core/dataframe/algebra.html#binary-operator).

The issue @Garra1980 mentioned is reproduced with Modin version before [commit with adding cell-wise execution](https://github.com/modin-project/modin/commit/8f35ab57996d18f2c1f001d624682952853454cd).

The root cause of issue is operation `df['c'].fillna(df['b'])` is performed on separated `Series` objects without full knowledge about the dataframe. The same flow we can reproduce using pandas:

```python
import pandas as pd

df = pd.DataFrame({'a': ['a'], 'b': ['b'],}, index=['row1'])
df['c'] = pd.NA

ser_left = pd.Series([pd.NA], name="c")
ser_right = pd.Series(["b"], name="b")
df['c'] = ser_left.fillna(ser_right) # changes nothing
print(df)
```

So, probably we need to change the implementation of `fillna` for such case.
> The issue @Garra1980 mentioned is reproduced with Modin version before [commit with adding cell-wise execution](https://github.com/modin-project/modin/commit/8f35ab57996d18f2c1f001d624682952853454cd).

Yes, there is another problem that I know how to fixed. [The problem](https://github.com/modin-project/modin/blob/a7354c9ca76525a265da98f2afe882c53f378840/modin/core/storage_formats/pandas/query_compiler.py#L1876) is that in our internal implementation we use single-column dataframes, which behave differently from the behavior of Series. Some example here:
```python
import pandas as pd

df = pd.DataFrame({'a': ['a'], 'b': ['b'],}, index=['row1'])
df['c'] = pd.NA
print(df['c'].fillna(df['b'])) # works as `df['c']` is a Series in pandas implementation
# attempt to emulate Modin internal behavior
print(df[['c']].fillna(df[['b']])) # all objects here are one-column Dataframes and this's not work
```
Your example isn't working because you need to specify index for Series:
```python
ser_left = pd.Series([pd.NA], name="c", index=["row1"])
ser_right = pd.Series(["b"], name="b", index=["row1"])
```

Apparently the only option left is to make reindexing optional.
> Your example isn't working because you need to specify index for Series:
> 
> ```python
> ser_left = pd.Series([pd.NA], name="c", index=["row1"])
> ser_right = pd.Series(["b"], name="b", index=["row1"])
> ```

Thanks for catch!

> Apparently the only option left is to make reindexing optional.

Could you please describe what "reindexing" do you mean?


> Could you please describe what "reindexing" do you mean?

`_copartition` calls `_join_index_objects`, which finds the common index to cast the left and right operand to, but for `fillna` doesn't need to do this (this is True for column reindexing). You can see my changes in https://github.com/modin-project/modin/pull/4674.

Commit message: remove new parameter from _copartition func

Signed-off-by: Myachev <anatoly.myachev@intel.com>

Patch to review:
diff --git a/docs/release_notes/release_notes-0.16.0.rst b/docs/release_notes/release_notes-0.16.0.rst
index 308c506b7bf..2e917e60a49 100644
--- a/docs/release_notes/release_notes-0.16.0.rst
+++ b/docs/release_notes/release_notes-0.16.0.rst
@@ -15,6 +15,7 @@ Key Features and Updates
   * FIX-#4604: Fix `groupby` + `agg` in case when multicolumn can arise (#4642)
   * FIX-#4582: Inherit custom log layer (#4583)
   * FIX-#4639: Fix `storage_options` usage for `read_csv` and `read_csv_glob` (#4644)
+  * FIX-#4660: Fix `fillna` when Modin series object is an argument (#4674)
   * FIX-#4593: Ensure Modin warns when setting columns via attributes (#4621)
   * FIX-#4584: Enable pdb debug when running cloud tests (#4585)
   * FIX-#4564: Workaround import issues in Ray: auto-import pandas on python start if env var is set (#4603)
diff --git a/modin/core/dataframe/pandas/dataframe/dataframe.py b/modin/core/dataframe/pandas/dataframe/dataframe.py
index 5b60d54191c..0932d4597cb 100644
--- a/modin/core/dataframe/pandas/dataframe/dataframe.py
+++ b/modin/core/dataframe/pandas/dataframe/dataframe.py
@@ -2646,7 +2646,9 @@ def _copartition(self, axis, other, how, sort, force_repartition=False):
         return (reindexed_frames[0], reindexed_frames[1:], joined_index, base_lengths)
 
     @lazy_metadata_decorator(apply_axis="both")
-    def n_ary_op(self, op, right_frames: list, join_type="outer"):
+    def n_ary_op(
+        self, op, right_frames: list, join_type="outer", perform_column_reindex=True
+    ):
         """
         Perform an n-opary operation by joining with other Modin DataFrame(s).
 
@@ -2658,6 +2660,8 @@ def n_ary_op(self, op, right_frames: list, join_type="outer"):
             Modin DataFrames to join with.
         join_type : str, default: "outer"
             Type of join to apply.
+        perform_column_reindex : bool, default: True
+            For some ops, for example `fillna`, column reindexing of right operand isn't needed.
 
         Returns
         -------
@@ -2667,26 +2671,35 @@ def n_ary_op(self, op, right_frames: list, join_type="outer"):
         left_parts, list_of_right_parts, joined_index, row_lengths = self._copartition(
             0, right_frames, join_type, sort=True
         )
-        new_left_frame = self.__constructor__(
-            left_parts, joined_index, self.columns, row_lengths, self.column_widths
-        )
-        new_right_frames = [
-            self.__constructor__(
-                right_parts,
-                joined_index,
-                right_frame.columns,
-                row_lengths,
-                right_frame.column_widths,
+        if perform_column_reindex:
+            new_left_frame = self.__constructor__(
+                left_parts, joined_index, self.columns, row_lengths, self.column_widths
             )
-            for right_parts, right_frame in zip(list_of_right_parts, right_frames)
-        ]
+            new_right_frames = [
+                self.__constructor__(
+                    right_parts,
+                    joined_index,
+                    right_frame.columns,
+                    row_lengths,
+                    right_frame.column_widths,
+                )
+                for right_parts, right_frame in zip(list_of_right_parts, right_frames)
+            ]
 
-        (
-            left_parts,
-            list_of_right_parts,
-            joined_columns,
-            column_widths,
-        ) = new_left_frame._copartition(1, new_right_frames, join_type, sort=True)
+            (
+                left_parts,
+                list_of_right_parts,
+                joined_columns,
+                column_widths,
+            ) = new_left_frame._copartition(
+                1,
+                new_right_frames,
+                join_type,
+                sort=True,
+            )
+        else:
+            joined_columns = self._columns_cache
+            column_widths = self._column_widths_cache
 
         new_frame = (
             np.array([])
diff --git a/modin/core/storage_formats/pandas/query_compiler.py b/modin/core/storage_formats/pandas/query_compiler.py
index 254e4de9145..495625297e3 100644
--- a/modin/core/storage_formats/pandas/query_compiler.py
+++ b/modin/core/storage_formats/pandas/query_compiler.py
@@ -1911,15 +1911,17 @@ def fillna_builder(series):
                     )
                 else:
 
-                    def fillna_builder(series, value_arg):
-                        # Both arguments for this function are 1-column `DataFrames` which denote `Series` type.
-                        # Because they are both of the same type, it is not necessary to convert either of them into
-                        # `Series` by squeezing since `fillna` works perfectly in the same way on 1-column `DataFrame`
-                        # objects (when `limit` parameter is absent) as it works on two `Series`.
-                        return series.fillna(value=value_arg, **kwargs)
+                    def fillna_builder(df, value_arg):
+                        if isinstance(value_arg, pandas.DataFrame):
+                            value_arg = value_arg.squeeze(axis=1)
+                        res = df.squeeze(axis=1).fillna(value=value_arg, **kwargs)
+                        return pandas.DataFrame(res)
 
                     new_modin_frame = self._modin_frame.n_ary_op(
-                        fillna_builder, [value._modin_frame], join_type="left"
+                        fillna_builder,
+                        [value._modin_frame],
+                        join_type="left",
+                        perform_column_reindex=False,
                     )
 
                 return self.__constructor__(new_modin_frame)
diff --git a/modin/pandas/test/dataframe/test_window.py b/modin/pandas/test/dataframe/test_window.py
index f1f7e103bbd..930b805ef89 100644
--- a/modin/pandas/test/dataframe/test_window.py
+++ b/modin/pandas/test/dataframe/test_window.py
@@ -206,6 +206,13 @@ def test_fillna_downcast():
     df_equals(modin_df, result)
 
 
+def test_fillna_4660():
+    eval_general(
+        *create_test_dfs({"a": ["a"], "b": ["b"], "c": [pd.NA]}, index=["row1"]),
+        lambda df: df["c"].fillna(df["b"]),
+    )
+
+
 def test_fillna_inplace():
     frame_data = random_state.randn(10, 4)
     df = pandas.DataFrame(frame_data)


Code review comments:
diff_hunk: @@ -2658,6 +2660,8 @@ def n_ary_op(self, op, right_frames: list, join_type="outer"):
             Modin DataFrames to join with.
         join_type : str, default: "outer"
             Type of join to apply.
+        perform_column_reindex : bool, default: True
+            For some ops, for example `fillna`, column reindexing of right operand isn't needed.
path: modin/core/dataframe/pandas/dataframe/dataframe.py
text: @user1:
```suggestion
            Whether to perform copartitioning along columns or not.
            For some ops this isn't needed (e.g., `fillna`).
```

diff_hunk: @@ -2667,26 +2671,35 @@ def n_ary_op(self, op, right_frames: list, join_type="outer"):
         left_parts, list_of_right_parts, joined_index, row_lengths = self._copartition(
             0, right_frames, join_type, sort=True
         )
-        new_left_frame = self.__constructor__(
-            left_parts, joined_index, self.columns, row_lengths, self.column_widths
-        )
-        new_right_frames = [
-            self.__constructor__(
-                right_parts,
-                joined_index,
-                right_frame.columns,
-                row_lengths,
-                right_frame.column_widths,
+        if perform_column_reindex:
path: modin/core/dataframe/pandas/dataframe/dataframe.py
text: @user1:
```suggestion
        if copartition_along_columns:
```

diff_hunk: @@ -2658,6 +2660,8 @@ def n_ary_op(self, op, right_frames: list, join_type="outer"):
             Modin DataFrames to join with.
         join_type : str, default: "outer"
             Type of join to apply.
+        perform_column_reindex : bool, default: True
path: modin/core/dataframe/pandas/dataframe/dataframe.py
text: @user1:
```suggestion
        copartition_along_columns : bool, default: True
```

diff_hunk: @@ -2646,7 +2646,9 @@ def _copartition(self, axis, other, how, sort, force_repartition=False):
         return (reindexed_frames[0], reindexed_frames[1:], joined_index, base_lengths)
 
     @lazy_metadata_decorator(apply_axis="both")
-    def n_ary_op(self, op, right_frames: list, join_type="outer"):
+    def n_ary_op(
+        self, op, right_frames: list, join_type="outer", perform_column_reindex=True
path: modin/core/dataframe/pandas/dataframe/dataframe.py
text: @user1:
```suggestion
        self, op, right_frames: list, join_type="outer", copartition_along_columns=True
```