Repository: modin-project/modin (python)

Issue:
BUG: Series binary operators don't respect fill_value
### Modin version checks

- [X] I have checked that this issue has not already been reported.

- [X] I have confirmed this bug exists on the latest released version of Modin.

- [X] I have confirmed this bug exists on the main branch of Modin. (In order to do this you can follow [this guide](https://modin.readthedocs.io/en/stable/getting_started/installation.html#installing-from-the-github-main-branch).)


### Reproducible Example

```python
import modin.pandas as pd
import numpy as np
pd.Series([13, 56, 4.5, np.nan, 0.99]).sub(pd.Series([55, np.nan]), fill_value=9.9)
```


### Issue Description

Modin ignores `fill_value` for most binary operators (with the exception of `and`), so operations like the above proceed with missing values.
```
>>> pd.Series([13, 56, 4.5, np.nan, 0.99]).sub(pd.Series([55, np.nan]), fill_value=9.9)
0   -42.0
1     NaN
2     NaN
3     NaN
4     NaN
dtype: float64
```

### Expected Behavior

pandas reference output:
```
>>> import pandas
>>> pandas.Series([13, 56, 4.5, np.nan, 0.99]).sub(pandas.Series([55, np.nan]), fill_value=9.9)
0   -42.00
1    46.10
2    -5.40
3      NaN
4    -8.91
dtype: float64
```

### Error Logs

<details>

```python-traceback

Replace this line with the error backtrace (if applicable).

```

</details>


### Installed Versions

<details>

INSTALLED VERSIONS
------------------
commit                : 2873676be1cef686456e0358c483d9e7bc49715c
python                : 3.10.13.final.0
python-bits           : 64
OS                    : Darwin
OS-release            : 23.6.0
Version               : Darwin Kernel Version 23.6.0: Mon Jul 29 21:13:04 PDT 2024; root:xnu-10063.141.2~1/RELEASE_ARM64_T6020
machine               : arm64
processor             : arm
byteorder             : little
LC_ALL                : None
LANG                  : en_US.UTF-8
LOCALE                : en_US.UTF-8

Modin dependencies
------------------
modin                 : 0.31.0+18.g2873676b
ray                   : 2.34.0
dask                  : 2024.8.1
distributed           : 2024.8.1

pandas dependencies
-------------------
pandas                : 2.2.2
numpy                 : 1.26.4
pytz                  : 2023.3.post1
dateutil              : 2.8.2
setuptools            : 68.0.0
pip                   : 23.3
Cython                : None
pytest                : 8.3.2
hypothesis            : None
sphinx                : 5.3.0
blosc                 : None
feather               : None
xlsxwriter            : None
lxml.etree            : 5.3.0
html5lib              : None
pymysql               : None
psycopg2              : 2.9.9
jinja2                : 3.1.4
IPython               : 8.17.2
pandas_datareader     : None
adbc-driver-postgresql: None
adbc-driver-sqlite    : None
bs4                   : 4.12.2
bottleneck            : None
dataframe-api-compat  : None
fastparquet           : 2024.5.0
fsspec                : 2024.6.1
gcsfs                 : None
matplotlib            : 3.9.2
numba                 : None
numexpr               : 2.10.1
odfpy                 : None
openpyxl              : 3.1.5
pandas_gbq            : 0.23.1
pyarrow               : 17.0.0
pyreadstat            : None
python-calamine       : None
pyxlsb                : None
s3fs                  : 2024.6.1
scipy                 : 1.14.1
sqlalchemy            : 2.0.32
tables                : 3.10.1
tabulate              : None
xarray                : 2024.7.0
xlrd                  : 2.0.1
zstandard             : None
tzdata                : 2023.3
qtpy                  : None
pyqt5                 : None

</details>


Pull request description:
FIX-#7381: Fix Series binary operators ignoring fill_value
<!--
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?

Many binary operators in the frontend Series class were not passing their `fillna` parameters to the query compiler (not sure why this wasn't caught in linting). For arithmetic functions, the fix is simply to propagate `fillna`, but for logical operators (`eq`/`ge` etc.), the equivalent DataFrame method does not take a `fillna` parameter, so we need to add a separate query compiler method to handle this case.

- [x] first commit message and PR title follow format outlined [here](https://modin.readthedocs.io/en/latest/development/contributing.html#commit-message-formatting)
  > **_NOTE:_**  If you edit the PR title to match this format, you need to add another commit (even if it's empty) or amend your last commit for the CI job that checks the PR title to pick up the new PR title.
- [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 #7381 <!-- 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 -->


Patch:
diff --git a/modin/core/storage_formats/base/doc_utils.py b/modin/core/storage_formats/base/doc_utils.py
index ce0f3935..6ba7a98a 100644
--- a/modin/core/storage_formats/base/doc_utils.py
+++ b/modin/core/storage_formats/base/doc_utils.py
@@ -193,6 +193,14 @@ def doc_binary_method(operation, sign, self_on_right=False, op_type="arithmetic"
         fill_value : float or None
             Value to fill missing elements during frame alignment.
         """,
+        "series_comparison": """
+        level : int or label
+            In case of MultiIndex match index values on the passed level.
+        fill_value : float or None
+            Value to fill missing elements during frame alignment.
+        axis : {{0, 1}}
+            Unused. Parameter needed for compatibility with DataFrame.
+        """,
     }
 
     verbose_substitution = (
diff --git a/modin/core/storage_formats/base/query_compiler.py b/modin/core/storage_formats/base/query_compiler.py
index 343008d2..965fb98e 100644
--- a/modin/core/storage_formats/base/query_compiler.py
+++ b/modin/core/storage_formats/base/query_compiler.py
@@ -647,6 +647,18 @@ class BaseQueryCompiler(
     def eq(self, other, **kwargs):  # noqa: PR02
         return BinaryDefault.register(pandas.DataFrame.eq)(self, other=other, **kwargs)
 
+    @doc_utils.doc_binary_method(
+        operation="equality comparison", sign="==", op_type="series_comparison"
+    )
+    def series_eq(self, other, **kwargs):  # noqa: PR02
+        return BinaryDefault.register(pandas.Series.eq)(
+            self,
+            other=other,
+            squeeze_self=True,
+            squeeze_other=kwargs.pop("squeeze_other", False),
+            **kwargs,
+        )
+
     @doc_utils.add_refer_to("DataFrame.equals")
     def equals(self, other):  # noqa: PR01, RT01
         return BinaryDefault.register(pandas.DataFrame.equals)(self, other=other)
@@ -685,24 +697,76 @@ class BaseQueryCompiler(
     def ge(self, other, **kwargs):  # noqa: PR02
         return BinaryDefault.register(pandas.DataFrame.ge)(self, other=other, **kwargs)
 
+    @doc_utils.doc_binary_method(
+        operation="greater than or equal comparison",
+        sign=">=",
+        op_type="series_comparison",
+    )
+    def series_ge(self, other, **kwargs):  # noqa: PR02
+        return BinaryDefault.register(pandas.Series.ge)(
+            self,
+            other=other,
+            squeeze_self=True,
+            squeeze_other=kwargs.pop("squeeze_other", False),
+            **kwargs,
+        )
+
     @doc_utils.doc_binary_method(
         operation="greater than comparison", sign=">", op_type="comparison"
     )
     def gt(self, other, **kwargs):  # noqa: PR02
         return BinaryDefault.register(pandas.DataFrame.gt)(self, other=other, **kwargs)
 
+    @doc_utils.doc_binary_method(
+        operation="greater than comparison", sign=">", op_type="series_comparison"
+    )
+    def series_gt(self, other, **kwargs):  # noqa: PR02
+        return BinaryDefault.register(pandas.Series.gt)(
+            self,
+            other=other,
+            squeeze_self=True,
+            squeeze_other=kwargs.pop("squeeze_other", False),
+            **kwargs,
+        )
+
     @doc_utils.doc_binary_method(
         operation="less than or equal comparison", sign="<=", op_type="comparison"
     )
     def le(self, other, **kwargs):  # noqa: PR02
         return BinaryDefault.register(pandas.DataFrame.le)(self, other=other, **kwargs)
 
+    @doc_utils.doc_binary_method(
+        operation="less than or equal comparison",
+        sign="<=",
+        op_type="series_comparison",
+    )
+    def series_le(self, other, **kwargs):  # noqa: PR02
+        return BinaryDefault.register(pandas.Series.le)(
+            self,
+            other=other,
+            squeeze_self=True,
+            squeeze_other=kwargs.pop("squeeze_other", False),
+            **kwargs,
+        )
+
     @doc_utils.doc_binary_method(
         operation="less than comparison", sign="<", op_type="comparison"
     )
     def lt(self, other, **kwargs):  # noqa: PR02
         return BinaryDefault.register(pandas.DataFrame.lt)(self, other=other, **kwargs)
 
+    @doc_utils.doc_binary_method(
+        operation="less than", sign="<", op_type="series_comparison"
+    )
+    def series_lt(self, other, **kwargs):  # noqa: PR02
+        return BinaryDefault.register(pandas.Series.lt)(
+            self,
+            other=other,
+            squeeze_self=True,
+            squeeze_other=kwargs.pop("squeeze_other", False),
+            **kwargs,
+        )
+
     @doc_utils.doc_binary_method(operation="modulo", sign="%")
     def mod(self, other, **kwargs):  # noqa: PR02
         return BinaryDefault.register(pandas.DataFrame.mod)(self, other=other, **kwargs)
@@ -818,6 +882,18 @@ class BaseQueryCompiler(
     def ne(self, other, **kwargs):  # noqa: PR02
         return BinaryDefault.register(pandas.DataFrame.ne)(self, other=other, **kwargs)
 
+    @doc_utils.doc_binary_method(
+        operation="not equal comparison", sign="!=", op_type="series_comparison"
+    )
+    def series_ne(self, other, **kwargs):  # noqa: PR02
+        return BinaryDefault.register(pandas.Series.ne)(
+            self,
+            other=other,
+            squeeze_self=True,
+            squeeze_other=kwargs.pop("squeeze_other", False),
+            **kwargs,
+        )
+
     @doc_utils.doc_binary_method(operation="exponential power", sign="**")
     def pow(self, other, **kwargs):  # noqa: PR02
         return BinaryDefault.register(pandas.DataFrame.pow)(self, other=other, **kwargs)
diff --git a/modin/core/storage_formats/pandas/query_compiler.py b/modin/core/storage_formats/pandas/query_compiler.py
index c7fb0bae..655d427d 100644
--- a/modin/core/storage_formats/pandas/query_compiler.py
+++ b/modin/core/storage_formats/pandas/query_compiler.py
@@ -253,6 +253,26 @@ def copy_df_for_func(func, display_name: str = None):
     return caller
 
 
+def _series_logical_binop(func):
+    """
+    Build a callable function to pass to Binary.register for Series logical operators.
+
+    Parameters
+    ----------
+    func : callable
+        Binary operator method of pandas.Series to be applied.
+
+    Returns
+    -------
+    callable
+    """
+    return lambda x, y, **kwargs: func(
+        x.squeeze(axis=1),
+        y.squeeze(axis=1) if kwargs.pop("squeeze_other", False) else y,
+        **kwargs,
+    ).to_frame()
+
+
 @_inherit_docstrings(BaseQueryCompiler)
 class PandasQueryCompiler(BaseQueryCompiler, QueryCompilerCaster):
     """
@@ -522,6 +542,26 @@ class PandasQueryCompiler(BaseQueryCompiler, QueryCompilerCaster):
         sort=False,
     )
 
+    # Series logical operators take an additional fill_value flag that dataframe does not
+    series_eq = Binary.register(
+        _series_logical_binop(pandas.Series.eq), infer_dtypes="bool"
+    )
+    series_ge = Binary.register(
+        _series_logical_binop(pandas.Series.ge), infer_dtypes="bool"
+    )
+    series_gt = Binary.register(
+        _series_logical_binop(pandas.Series.gt), infer_dtypes="bool"
+    )
+    series_le = Binary.register(
+        _series_logical_binop(pandas.Series.le), infer_dtypes="bool"
+    )
+    series_lt = Binary.register(
+        _series_logical_binop(pandas.Series.lt), infer_dtypes="bool"
+    )
+    series_ne = Binary.register(
+        _series_logical_binop(pandas.Series.ne), infer_dtypes="bool"
+    )
+
     # Needed for numpy API
     _logical_and = Binary.register(
         lambda df, other, *args, **kwargs: pandas.DataFrame(
diff --git a/modin/pandas/base.py b/modin/pandas/base.py
index 04dd8459..1b893743 100644
--- a/modin/pandas/base.py
+++ b/modin/pandas/base.py
@@ -509,6 +509,17 @@ class BasePandasDataset(ClassLogger):
         ]
         if op in exclude_list:
             kwargs.pop("axis")
+        # Series logical operations take an additional fill_value argument that DF does not
+        series_specialize_list = [
+            "eq",
+            "ge",
+            "gt",
+            "le",
+            "lt",
+            "ne",
+        ]
+        if not self._is_dataframe and op in series_specialize_list:
+            op = "series_" + op
         new_query_compiler = getattr(self._query_compiler, op)(other, **kwargs)
         return self._create_or_update_from_compiler(new_query_compiler)
 
diff --git a/modin/pandas/series.py b/modin/pandas/series.py
index d18a0bec..20825343 100644
--- a/modin/pandas/series.py
+++ b/modin/pandas/series.py
@@ -1022,7 +1022,12 @@ class Series(BasePandasDataset):
         """
         Indicate duplicate Series values.
         """
-        return self.to_frame().duplicated(keep=keep)
+        name = self.name
+        result = self.to_frame().duplicated(keep=keep)
+        # DataFrame.duplicated drops the name, so we need to manually restore it
+        if name is not None:
+            result.name = name
+        return result
 
     def eq(
         self, other, level=None, fill_value=None, axis=0
@@ -1031,7 +1036,14 @@ class Series(BasePandasDataset):
         Return Equal to of series and `other`, element-wise (binary operator `eq`).
         """
         new_self, new_other = self._prepare_inter_op(other)
-        return super(Series, new_self).eq(new_other, level=level, axis=axis)
+        return new_self._binary_op(
+            "eq",
+            new_other,
+            level=level,
+            fill_value=fill_value,
+            axis=axis,
+            squeeze_other=isinstance(other, Series),
+        )
 
     def equals(self, other) -> bool:  # noqa: PR01, RT01, D200
         """
@@ -1130,7 +1142,7 @@ class Series(BasePandasDataset):
         """
         new_self, new_other = self._prepare_inter_op(other)
         return super(Series, new_self).floordiv(
-            new_other, level=level, fill_value=None, axis=axis
+            new_other, level=level, fill_value=fill_value, axis=axis
         )
 
     def ge(
@@ -1140,7 +1152,14 @@ class Series(BasePandasDataset):
         Return greater than or equal to of series and `other`, element-wise (binary operator `ge`).
         """
         new_self, new_other = self._prepare_inter_op(other)
-        return super(Series, new_self).ge(new_other, level=level, axis=axis)
+        return new_self._binary_op(
+            "ge",
+            new_other,
+            level=level,
+            fill_value=fill_value,
+            axis=axis,
+            squeeze_other=isinstance(other, Series),
+        )
 
     def groupby(
         self,
@@ -1188,7 +1207,14 @@ class Series(BasePandasDataset):
         Return greater than of series and `other`, element-wise (binary operator `gt`).
         """
         new_self, new_other = self._prepare_inter_op(other)
-        return super(Series, new_self).gt(new_other, level=level, axis=axis)
+        return new_self._binary_op(
+            "gt",
+            new_other,
+            level=level,
+            fill_value=fill_value,
+            axis=axis,
+            squeeze_other=isinstance(other, Series),
+        )
 
     def hist(
         self,
@@ -1306,7 +1332,14 @@ class Series(BasePandasDataset):
         Return less than or equal to of series and `other`, element-wise (binary operator `le`).
         """
         new_self, new_other = self._prepare_inter_op(other)
-        return super(Series, new_self).le(new_other, level=level, axis=axis)
+        return new_self._binary_op(
+            "le",
+            new_other,
+            level=level,
+            fill_value=fill_value,
+            axis=axis,
+            squeeze_other=isinstance(other, Series),
+        )
 
     def lt(
         self, other, level=None, fill_value=None, axis=0
@@ -1315,7 +1348,14 @@ class Series(BasePandasDataset):
         Return less than of series and `other`, element-wise (binary operator `lt`).
         """
         new_self, new_other = self._prepare_inter_op(other)
-        return super(Series, new_self).lt(new_other, level=level, axis=axis)
+        return new_self._binary_op(
+            "lt",
+            new_other,
+            level=level,
+            fill_value=fill_value,
+            axis=axis,
+            squeeze_other=isinstance(other, Series),
+        )
 
     def map(self, arg, na_action=None) -> Series:  # noqa: PR01, RT01, D200
         """
@@ -1402,7 +1442,7 @@ class Series(BasePandasDataset):
         """
         new_self, new_other = self._prepare_inter_op(other)
         return super(Series, new_self).mod(
-            new_other, level=level, fill_value=None, axis=axis
+            new_other, level=level, fill_value=fill_value, axis=axis
         )
 
     def mode(self, dropna=True) -> Series:  # noqa: PR01, RT01, D200
@@ -1419,7 +1459,7 @@ class Series(BasePandasDataset):
         """
         new_self, new_other = self._prepare_inter_op(other)
         return super(Series, new_self).mul(
-            new_other, level=level, fill_value=None, axis=axis
+            new_other, level=level, fill_value=fill_value, axis=axis
         )
 
     multiply = mul
@@ -1432,7 +1472,7 @@ class Series(BasePandasDataset):
         """
         new_self, new_other = self._prepare_inter_op(other)
         return super(Series, new_self).rmul(
-            new_other, level=level, fill_value=None, axis=axis
+            new_other, level=level, fill_value=fill_value, axis=axis
         )
 
     def ne(
@@ -1442,7 +1482,14 @@ class Series(BasePandasDataset):
         Return not equal to of series and `other`, element-wise (binary operator `ne`).
         """
         new_self, new_other = self._prepare_inter_op(other)
-        return super(Series, new_self).ne(new_other, level=level, axis=axis)
+        return new_self._binary_op(
+            "ne",
+            new_other,
+            level=level,
+            fill_value=fill_value,
+            axis=axis,
+            squeeze_other=isinstance(other, Series),
+        )
 
     def nlargest(self, n=5, keep="first") -> Series:  # noqa: PR01, RT01, D200
         """
@@ -1562,7 +1609,7 @@ class Series(BasePandasDataset):
         """
         new_self, new_other = self._prepare_inter_op(other)
         return super(Series, new_self).pow(
-            new_other, level=level, fill_value=None, axis=axis
+            new_other, level=level, fill_value=fill_value, axis=axis
         )
 
     @_inherit_docstrings(pandas.Series.prod, apilink="pandas.Series.prod")
@@ -1763,7 +1810,7 @@ class Series(BasePandasDataset):
         """
         new_self, new_other = self._prepare_inter_op(other)
         return super(Series, new_self).rfloordiv(
-            new_other, level=level, fill_value=None, axis=axis
+            new_other, level=level, fill_value=fill_value, axis=axis
         )
 
     def rmod(
@@ -1774,7 +1821,7 @@ class Series(BasePandasDataset):
         """
         new_self, new_other = self._prepare_inter_op(other)
         return super(Series, new_self).rmod(
-            new_other, level=level, fill_value=None, axis=axis
+            new_other, level=level, fill_value=fill_value, axis=axis
         )
 
     def rpow(
@@ -1785,7 +1832,7 @@ class Series(BasePandasDataset):
         """
         new_self, new_other = self._prepare_inter_op(other)
         return super(Series, new_self).rpow(
-            new_other, level=level, fill_value=None, axis=axis
+            new_other, level=level, fill_value=fill_value, axis=axis
         )
 
     def rsub(
@@ -1796,7 +1843,7 @@ class Series(BasePandasDataset):
         """
         new_self, new_other = self._prepare_inter_op(other)
         return super(Series, new_self).rsub(
-            new_other, level=level, fill_value=None, axis=axis
+            new_other, level=level, fill_value=fill_value, axis=axis
         )
 
     def rtruediv(
@@ -1807,7 +1854,7 @@ class Series(BasePandasDataset):
         """
         new_self, new_other = self._prepare_inter_op(other)
         return super(Series, new_self).rtruediv(
-            new_other, level=level, fill_value=None, axis=axis
+            new_other, level=level, fill_value=fill_value, axis=axis
         )
 
     rdiv = rtruediv
@@ -1955,7 +2002,7 @@ class Series(BasePandasDataset):
         """
         new_self, new_other = self._prepare_inter_op(other)
         return super(Series, new_self).sub(
-            new_other, level=level, fill_value=None, axis=axis
+            new_other, level=level, fill_value=fill_value, axis=axis
         )
 
     subtract = sub
@@ -2130,7 +2177,7 @@ class Series(BasePandasDataset):
         """
         new_self, new_other = self._prepare_inter_op(other)
         return super(Series, new_self).truediv(
-            new_other, level=level, fill_value=None, axis=axis
+            new_other, level=level, fill_value=fill_value, axis=axis
         )
 
     div = divide = truediv


Test patch:
diff --git a/modin/tests/pandas/test_series.py b/modin/tests/pandas/test_series.py
index 9dd8b98a..115b6679 100644
--- a/modin/tests/pandas/test_series.py
+++ b/modin/tests/pandas/test_series.py
@@ -1942,6 +1942,12 @@ def test_duplicated(data, keep):
     df_equals(modin_result, pandas_series.duplicated(keep=keep))
 
 
+def test_duplicated_keeps_name_issue_7375():
+    # Ensure that the name property of a series is preserved across duplicated
+    modin_series, pandas_series = create_test_series([1, 2, 3, 1], name="a")
+    df_equals(modin_series.duplicated(), pandas_series.duplicated())
+
+
 @pytest.mark.parametrize("data", test_data_values, ids=test_data_keys)
 def test_empty(data):
     modin_series, pandas_series = create_test_series(data)
@@ -5094,3 +5100,48 @@ def test__reduce__():
         .rename("league")
     )
     df_equals(result_md, result_pd)
+
+
+@pytest.mark.parametrize(
+    "op",
+    [
+        "add",
+        "radd",
+        "divmod",
+        "eq",
+        "floordiv",
+        "ge",
+        "gt",
+        "le",
+        "lt",
+        "mod",
+        "mul",
+        "rmul",
+        "ne",
+        "pow",
+        "rdivmod",
+        "rfloordiv",
+        "rmod",
+        "rpow",
+        "rsub",
+        "rtruediv",
+        "sub",
+        "truediv",
+    ],
+)
+def test_binary_with_fill_value_issue_7381(op):
+    # Ensures that series binary operations respect the fill_value flag
+    series_md, series_pd = create_test_series([0, 1, 2, 3])
+    rhs_md, rhs_pd = create_test_series([0])
+    result_md = getattr(series_md, op)(rhs_md, fill_value=2)
+    result_pd = getattr(series_pd, op)(rhs_pd, fill_value=2)
+    df_equals(result_md, result_pd)
+
+
+@pytest.mark.parametrize("op", ["eq", "ge", "gt", "le", "lt", "ne"])
+def test_logical_binary_with_list(op):
+    series_md, series_pd = create_test_series([0, 1, 2])
+    rhs = [2, 1, 0]
+    result_md = getattr(series_md, op)(rhs)
+    result_pd = getattr(series_pd, op)(rhs)
+    df_equals(result_md, result_pd)