Repository: modin-project/modin
Pull request: PERF-#5182: Precompute dtypes when performing binary operations in certain cases
<!--
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?

Added logic to precompute dtypes before add operation was performed on two dataframes.

Benchmarking code

```
from modin.pandas import DataFrame
import modin.pandas as pd
import numpy as np
import time
rows=1000
cols=100000
df = DataFrame(np.ones((rows, cols)),dtype=np.int32)
df1=df.copy()

start_time = time.time()
df_sum = df1 / df
df_sum.astype('int32')
print("--- %s seconds ---" % (time.time() - start_time))
```


For master branch

**--- 20.01868987083435 seconds ---**

For the current change

**--- 4.792952299118042 seconds ---**



For boolean operators  (Replacing `df_sum = df1 / df` with below line )


`df_sum = df1 > df
`


For Master:
**--- 21.683951139450073 seconds ---**

For current

**--- 5.464660882949829 seconds ---**

<!-- Please give a short brief about these changes. -->

- [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 #5182 ? <!-- 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 -->


Problem statement:
[PERF] compute `dtypes` when performing binary operations
It should be possible and beneficial to compute `dtypes` when doing binary operations without remote calls so other subsequent operations, which require `dypes` to be known, do not perform the remote calls to compute it.

Here are some examples when we can compute/pass known dtypes to the core dataframe.

1. `left_df + right_df`, leftleft_df.dtypes and right_df.dtypes are known and equal to each other
2. `left_df + 1`, left_df.dtypes is known, dtype of 1 is int.

Hints from issue discussion:
https://github.com/modin-project/modin/issues/4868
As resultant dtypes for each function  are dependent on the operation. Looks like this has to be done specific to each function. 

eg For add(df1+d2)   for each column the resultant dtype would be determined by  any of the 2 below cases



**Case 1**)Check whether dtype of column in df1/df2 can be casted to dtype  of column in df2/df1.  eg If np.int8  can be casted to np.int16, then np.int16 becomes resultant dtype  

eg 

```
df1 = pd.DataFrame(np.ones((10, 10)),dtype=np.int8)
df2 = pd.DataFrame(np.ones((10, 10)),dtype=np.int16)
df3=df1+df2
print(df3.dtypes)  # resultant would be of dtype int16 
```

**Case 2**)In case if dtypes cant be casted. A dtype which is just higher than both of the dtypes is assigned

eg:

```
df1 = pd.DataFrame(np.ones((10, 10)),dtype=np.float16)
df2 = pd.DataFrame(np.ones((10, 10)),dtype=np.int16)
df3=df1+df2
print(df3.dtypes)  # prints float32 which is not in any of the input dtypes
```

Note: With pandas supporting 20 dtypes there can be 190 ways to combine them. With 163 of them falling in first case and 27 in second case. The first case is easily implementable. Looking into ways to implement case 2 without giving say 27 rules to handle each set of dtypes.
We could use the following method from pandas. It seems to me it is used in pandas to determine resultant dtypes.

```python
from pandas.core.dtypes.cast import find_common_type

find_common_type([np.float16, np.int16])
dtype('float32')
```

Commit message: removing dtypes calls in base.py

Signed-off-by: arunjose696 <arunjose696@gmail.com>

Patch to review:
diff --git a/modin/core/dataframe/algebra/binary.py b/modin/core/dataframe/algebra/binary.py
index 2d9645aeaca..be2964e38d9 100644
--- a/modin/core/dataframe/algebra/binary.py
+++ b/modin/core/dataframe/algebra/binary.py
@@ -19,11 +19,98 @@
 from .operator import Operator
 
 
+def coerce_int_to_float64(dtype: np.dtype) -> np.dtype:
+    """
+    Coerce dtype to float64 if it is a variant of integer.
+
+    If dtype is integer, function returns float64 datatype.
+    If not, returns the datatype argument itself.
+
+    Parameters
+    ----------
+    dtype : np.dtype
+        NumPy datatype.
+
+    Returns
+    -------
+    dtype : np.dtype
+        Returns float64 for all int datatypes or returns the datatype itself
+        for other types.
+
+    Notes
+    -----
+    Used to precompute datatype in case of division in pandas.
+    """
+    if dtype in np.sctypes["int"] + np.sctypes["uint"]:
+        return np.dtype(np.float64)
+    else:
+        return dtype
+
+
+def compute_dtypes_common_cast(first, second) -> np.dtype:
+    """
+    Precompute data types for binary operations by finding common type between operands.
+
+    Parameters
+    ----------
+    first : PandasQueryCompiler
+        First operand for which the binary operation would be performed later.
+    second : PandasQueryCompiler
+        Second operand for which the binary operation would be performed later.
+
+    Returns
+    -------
+    dtypes
+        The pandas series with precomputed dtypes.
+
+    Notes
+    -----
+    The dtypes of the operands are supposed to be known.
+    """
+    dtypes_first = dict(zip(first.columns, first._modin_frame._dtypes))
+    dtypes_second = dict(zip(second.columns, second._modin_frame._dtypes))
+    columns_first = set(first.columns)
+    columns_second = set(second.columns)
+    common_columns = columns_first.intersection(columns_second)
+    mismatch_columns = columns_first.union(columns_second) - common_columns
+    # If at least one column doesn't match, the result of the non matching column would be nan.
+    nan_dtype = np.dtype(type(np.nan))
+    dtypes = pandas.Series(
+        [
+            pandas.core.dtypes.cast.find_common_type(
+                [
+                    dtypes_first[x],
+                    dtypes_second[x],
+                ]
+            )
+            for x in common_columns
+        ],
+        index=common_columns,
+    )
+    dtypes = pandas.concat(
+        [
+            dtypes,
+            pandas.Series(
+                [nan_dtype] * (len(mismatch_columns)),
+                index=mismatch_columns,
+            ),
+        ]
+    )
+    dtypes = dtypes.sort_index()
+    return dtypes
+
+
 class Binary(Operator):
     """Builder class for Binary operator."""
 
     @classmethod
-    def register(cls, func, join_type="outer", labels="replace"):
+    def register(
+        cls,
+        func,
+        join_type="outer",
+        labels="replace",
+        how_compute_dtypes=None,
+    ):
         """
         Build template binary operator.
 
@@ -36,6 +123,14 @@ def register(cls, func, join_type="outer", labels="replace"):
         labels : {"keep", "replace", "drop"}, default: "replace"
             Whether keep labels from left Modin DataFrame, replace them with labels
             from joined DataFrame or drop altogether to make them be computed lazily later.
+        how_compute_dtypes : {"common_cast", "float","bool", None}, default: None
+            How dtypes should be inferred.
+                * If "common_cast", casts to common dtype of operand columns.
+                * If "float", performs type casting by finding common dtype.
+                   If the common dtype is any of the integer types, perform type casting to float.
+                   Used in case of truediv.
+                * If "bool", dtypes would be a boolean series with same size as that of operands.
+                * If ``None``, do not infer new dtypes (they will be computed manually once accessed).
 
         Returns
         -------
@@ -96,13 +191,28 @@ def caller(
                         )
                     )
                 else:
+                    if (
+                        other._modin_frame._dtypes is not None
+                        and query_compiler._modin_frame._dtypes is not None
+                    ):
+                        if how_compute_dtypes == "bool":
+                            dtypes = pandas.Series(
+                                [np.dtype(bool)] * len(other._modin_frame._dtypes)
+                            )
+                        if how_compute_dtypes == "common_cast":
+                            dtypes = compute_dtypes_common_cast(query_compiler, other)
+                        elif how_compute_dtypes == "float":
+                            dtypes = compute_dtypes_common_cast(query_compiler, other)
+                            dtypes = dtypes.apply(coerce_int_to_float64)
                     return query_compiler.__constructor__(
                         query_compiler._modin_frame.n_ary_op(
                             lambda x, y: func(x, y, *args, **kwargs),
                             [other._modin_frame],
                             join_type=join_type,
+                            dtypes=dtypes,
                         )
                     )
+
             else:
                 # TODO: it's possible to chunk the `other` and broadcast them to partitions
                 # accordingly, in that way we will be able to use more efficient `._modin_frame.map()`
diff --git a/modin/core/dataframe/pandas/dataframe/dataframe.py b/modin/core/dataframe/pandas/dataframe/dataframe.py
index 07067ef83f4..795888ab021 100644
--- a/modin/core/dataframe/pandas/dataframe/dataframe.py
+++ b/modin/core/dataframe/pandas/dataframe/dataframe.py
@@ -2815,7 +2815,12 @@ def _copartition(self, axis, other, how, sort, force_repartition=False):
 
     @lazy_metadata_decorator(apply_axis="both")
     def n_ary_op(
-        self, op, right_frames: list, join_type="outer", copartition_along_columns=True
+        self,
+        op,
+        right_frames: list,
+        join_type="outer",
+        copartition_along_columns=True,
+        dtypes=None,
     ):
         """
         Perform an n-opary operation by joining with other Modin DataFrame(s).
@@ -2831,6 +2836,9 @@ def n_ary_op(
         copartition_along_columns : bool, default: True
             Whether to perform copartitioning along columns or not.
             For some ops this isn't needed (e.g., `fillna`).
+        dtypes : series, default: None
+            Dtypes of the resultant dataframe, this argument will be
+            received if the resultant dtypes of n-opary operation is precomputed.
 
         Returns
         -------
@@ -2885,6 +2893,7 @@ def n_ary_op(
             joined_columns,
             row_lengths,
             column_widths,
+            dtypes,
         )
 
     @lazy_metadata_decorator(apply_axis="both")
diff --git a/modin/core/storage_formats/pandas/query_compiler.py b/modin/core/storage_formats/pandas/query_compiler.py
index 25e91401aa2..732a98a64a8 100644
--- a/modin/core/storage_formats/pandas/query_compiler.py
+++ b/modin/core/storage_formats/pandas/query_compiler.py
@@ -375,34 +375,42 @@ def to_numpy(self, **kwargs):
     # such that columns/rows that don't have an index on the other DataFrame
     # result in NaN values.
 
-    add = Binary.register(pandas.DataFrame.add)
-    combine = Binary.register(pandas.DataFrame.combine)
-    combine_first = Binary.register(pandas.DataFrame.combine_first)
-    eq = Binary.register(pandas.DataFrame.eq)
-    floordiv = Binary.register(pandas.DataFrame.floordiv)
-    ge = Binary.register(pandas.DataFrame.ge)
-    gt = Binary.register(pandas.DataFrame.gt)
-    le = Binary.register(pandas.DataFrame.le)
-    lt = Binary.register(pandas.DataFrame.lt)
-    mod = Binary.register(pandas.DataFrame.mod)
-    mul = Binary.register(pandas.DataFrame.mul)
-    rmul = Binary.register(pandas.DataFrame.rmul)
-    ne = Binary.register(pandas.DataFrame.ne)
-    pow = Binary.register(pandas.DataFrame.pow)
-    radd = Binary.register(pandas.DataFrame.radd)
-    rfloordiv = Binary.register(pandas.DataFrame.rfloordiv)
-    rmod = Binary.register(pandas.DataFrame.rmod)
-    rpow = Binary.register(pandas.DataFrame.rpow)
-    rsub = Binary.register(pandas.DataFrame.rsub)
-    rtruediv = Binary.register(pandas.DataFrame.rtruediv)
-    sub = Binary.register(pandas.DataFrame.sub)
-    truediv = Binary.register(pandas.DataFrame.truediv)
-    __and__ = Binary.register(pandas.DataFrame.__and__)
-    __or__ = Binary.register(pandas.DataFrame.__or__)
-    __rand__ = Binary.register(pandas.DataFrame.__rand__)
-    __ror__ = Binary.register(pandas.DataFrame.__ror__)
-    __rxor__ = Binary.register(pandas.DataFrame.__rxor__)
-    __xor__ = Binary.register(pandas.DataFrame.__xor__)
+    add = Binary.register(pandas.DataFrame.add, how_compute_dtypes="common_cast")
+    combine = Binary.register(
+        pandas.DataFrame.combine, how_compute_dtypes="common_cast"
+    )
+    combine_first = Binary.register(
+        pandas.DataFrame.combine_first, how_compute_dtypes="bool"
+    )
+    eq = Binary.register(pandas.DataFrame.eq, how_compute_dtypes="bool")
+    floordiv = Binary.register(
+        pandas.DataFrame.floordiv, how_compute_dtypes="common_cast"
+    )
+    ge = Binary.register(pandas.DataFrame.ge, how_compute_dtypes="bool")
+    gt = Binary.register(pandas.DataFrame.gt, how_compute_dtypes="bool")
+    le = Binary.register(pandas.DataFrame.le, how_compute_dtypes="bool")
+    lt = Binary.register(pandas.DataFrame.lt, how_compute_dtypes="bool")
+    mod = Binary.register(pandas.DataFrame.mod, how_compute_dtypes="common_cast")
+    mul = Binary.register(pandas.DataFrame.mul, how_compute_dtypes="common_cast")
+    rmul = Binary.register(pandas.DataFrame.rmul, how_compute_dtypes="common_cast")
+    ne = Binary.register(pandas.DataFrame.ne, how_compute_dtypes="bool")
+    pow = Binary.register(pandas.DataFrame.pow, how_compute_dtypes="common_cast")
+    radd = Binary.register(pandas.DataFrame.radd, how_compute_dtypes="common_cast")
+    rfloordiv = Binary.register(
+        pandas.DataFrame.rfloordiv, how_compute_dtypes="common_cast"
+    )
+    rmod = Binary.register(pandas.DataFrame.rmod, how_compute_dtypes="common_cast")
+    rpow = Binary.register(pandas.DataFrame.rpow, how_compute_dtypes="common_cast")
+    rsub = Binary.register(pandas.DataFrame.rsub, how_compute_dtypes="common_cast")
+    rtruediv = Binary.register(pandas.DataFrame.rtruediv, how_compute_dtypes="float")
+    sub = Binary.register(pandas.DataFrame.sub, how_compute_dtypes="common_cast")
+    truediv = Binary.register(pandas.DataFrame.truediv, how_compute_dtypes="float")
+    __and__ = Binary.register(pandas.DataFrame.__and__, how_compute_dtypes="bool")
+    __or__ = Binary.register(pandas.DataFrame.__or__, how_compute_dtypes="bool")
+    __rand__ = Binary.register(pandas.DataFrame.__rand__, how_compute_dtypes="bool")
+    __ror__ = Binary.register(pandas.DataFrame.__ror__, how_compute_dtypes="bool")
+    __rxor__ = Binary.register(pandas.DataFrame.__rxor__, how_compute_dtypes="bool")
+    __xor__ = Binary.register(pandas.DataFrame.__xor__, how_compute_dtypes="bool")
     df_update = Binary.register(
         copy_df_for_func(pandas.DataFrame.update, display_name="update"),
         join_type="left",
diff --git a/modin/pandas/base.py b/modin/pandas/base.py
index 0470182cb60..82a4f168801 100644
--- a/modin/pandas/base.py
+++ b/modin/pandas/base.py
@@ -1402,7 +1402,7 @@ def eq(self, other, axis="columns", level=None):  # noqa: PR01, RT01, D200
         """
         Get equality of `BasePandasDataset` and `other`, element-wise (binary operator `eq`).
         """
-        return self._binary_op("eq", other, axis=axis, level=level, dtypes=np.bool_)
+        return self._binary_op("eq", other, axis=axis, level=level)
 
     def explode(self, column, ignore_index: bool = False):  # noqa: PR01, RT01, D200
         """
@@ -1628,7 +1628,7 @@ def ge(self, other, axis="columns", level=None):  # noqa: PR01, RT01, D200
         """
         Get greater than or equal comparison of `BasePandasDataset` and `other`, element-wise (binary operator `ge`).
         """
-        return self._binary_op("ge", other, axis=axis, level=level, dtypes=np.bool_)
+        return self._binary_op("ge", other, axis=axis, level=level)
 
     def get(self, key, default=None):  # noqa: PR01, RT01, D200
         """
@@ -1644,7 +1644,7 @@ def gt(self, other, axis="columns", level=None):  # noqa: PR01, RT01, D200
         """
         Get greater than comparison of `BasePandasDataset` and `other`, element-wise (binary operator `gt`).
         """
-        return self._binary_op("gt", other, axis=axis, level=level, dtypes=np.bool_)
+        return self._binary_op("gt", other, axis=axis, level=level)
 
     def head(self, n=5):  # noqa: PR01, RT01, D200
         """
@@ -1793,13 +1793,13 @@ def le(self, other, axis="columns", level=None):  # noqa: PR01, RT01, D200
         """
         Get less than or equal comparison of `BasePandasDataset` and `other`, element-wise (binary operator `le`).
         """
-        return self._binary_op("le", other, axis=axis, level=level, dtypes=np.bool_)
+        return self._binary_op("le", other, axis=axis, level=level)
 
     def lt(self, other, axis="columns", level=None):  # noqa: PR01, RT01, D200
         """
         Get less than comparison of `BasePandasDataset` and `other`, element-wise (binary operator `lt`).
         """
-        return self._binary_op("lt", other, axis=axis, level=level, dtypes=np.bool_)
+        return self._binary_op("lt", other, axis=axis, level=level)
 
     @property
     def loc(self):  # noqa: RT01, D200
@@ -2040,7 +2040,7 @@ def ne(self, other, axis="columns", level=None):  # noqa: PR01, RT01, D200
         """
         Get Not equal comparison of `BasePandasDataset` and `other`, element-wise (binary operator `ne`).
         """
-        return self._binary_op("ne", other, axis=axis, level=level, dtypes=np.bool_)
+        return self._binary_op("ne", other, axis=axis, level=level)
 
     def notna(self):  # noqa: RT01, D200
         """


Code review comments:
diff_hunk: @@ -19,11 +19,98 @@
 from .operator import Operator
 
 
+def coerce_int_to_float64(dtype: np.dtype) -> np.dtype:
+    """
+    Coerce dtype to float64 if it is a variant of integer.
+
+    If dtype is integer, function returns float64 datatype.
+    If not, returns the datatype argument itself.
+
+    Parameters
+    ----------
+    dtype : np.dtype
+        NumPy datatype.
+
+    Returns
+    -------
+    dtype : np.dtype
+        Returns float64 for all int datatypes or returns the datatype itself
+        for other types.
+
+    Notes
+    -----
+    Used to precompute datatype in case of division in pandas.
+    """
+    if dtype in np.sctypes["int"] + np.sctypes["uint"]:
+        return np.dtype(np.float64)
+    else:
+        return dtype
+
+
+def compute_dtypes_common_cast(first, second) -> np.dtype:
+    """
+    Precompute data types for binary operations by finding common type between operands.
+
+    Parameters
+    ----------
+    first : PandasQueryCompiler
+        First operand for which the binary operation would be performed later.
+    second : PandasQueryCompiler
+        Second operand for which the binary operation would be performed later.
+
+    Returns
+    -------
+    dtypes
+        The pandas series with precomputed dtypes.
+
+    Notes
+    -----
+    The dtypes of the operands are supposed to be known.
+    """
+    dtypes_first = dict(zip(first.columns, first._modin_frame._dtypes))
+    dtypes_second = dict(zip(second.columns, second._modin_frame._dtypes))
path: modin/core/dataframe/algebra/binary.py
text: @user1:
```suggestion
    dtypes_first = first._modin_frame._dtypes.to_dict()
    dtypes_second = second._modin_frame._dtypes.to_dict()
```
or even we can remove this at all and use just this further.

```python
dtypes_first = first._modin_frame._dtypes
dtypes_first.loc[col]
```

@author:
I prefer to use .to_dict() method as in first suggestion. In the second approach again a list would have to be created to iterate over the col names with `dtypes_first.keys()`, which makes them both take up same space at end,
I feel the .to_dict() method to be more readable.

diff_hunk: @@ -19,11 +19,98 @@
 from .operator import Operator
 
 
+def coerce_int_to_float64(dtype: np.dtype) -> np.dtype:
+    """
+    Coerce dtype to float64 if it is a variant of integer.
+
+    If dtype is integer, function returns float64 datatype.
+    If not, returns the datatype argument itself.
+
+    Parameters
+    ----------
+    dtype : np.dtype
+        NumPy datatype.
+
+    Returns
+    -------
+    dtype : np.dtype
+        Returns float64 for all int datatypes or returns the datatype itself
+        for other types.
+
+    Notes
+    -----
+    Used to precompute datatype in case of division in pandas.
+    """
+    if dtype in np.sctypes["int"] + np.sctypes["uint"]:
+        return np.dtype(np.float64)
+    else:
+        return dtype
+
+
+def compute_dtypes_common_cast(first, second) -> np.dtype:
+    """
+    Precompute data types for binary operations by finding common type between operands.
+
+    Parameters
+    ----------
+    first : PandasQueryCompiler
+        First operand for which the binary operation would be performed later.
+    second : PandasQueryCompiler
+        Second operand for which the binary operation would be performed later.
+
+    Returns
+    -------
+    dtypes
+        The pandas series with precomputed dtypes.
+
+    Notes
+    -----
+    The dtypes of the operands are supposed to be known.
+    """
+    dtypes_first = dict(zip(first.columns, first._modin_frame._dtypes))
+    dtypes_second = dict(zip(second.columns, second._modin_frame._dtypes))
+    columns_first = set(first.columns)
+    columns_second = set(second.columns)
+    common_columns = columns_first.intersection(columns_second)
+    mismatch_columns = columns_first.union(columns_second) - common_columns
+    # If at least one column doesn't match, the result of the non matching column would be nan.
+    nan_dtype = np.dtype(type(np.nan))
+    dtypes = pandas.Series(
+        [
+            pandas.core.dtypes.cast.find_common_type(
+                [
+                    dtypes_first[x],
+                    dtypes_second[x],
+                ]
+            )
+            for x in common_columns
+        ],
+        index=common_columns,
+    )
+    dtypes = pandas.concat(
+        [
+            dtypes,
+            pandas.Series(
+                [nan_dtype] * (len(mismatch_columns)),
+                index=mismatch_columns,
+            ),
+        ]
+    )
+    dtypes = dtypes.sort_index()
+    return dtypes
+
+
 class Binary(Operator):
     """Builder class for Binary operator."""
 
     @classmethod
-    def register(cls, func, join_type="outer", labels="replace"):
+    def register(
+        cls,
+        func,
+        join_type="outer",
+        labels="replace",
+        how_compute_dtypes=None,
path: modin/core/dataframe/algebra/binary.py
text: @user1:
```suggestion
        infer_dtypes=None,
```
I suggest renaming this to `infer_dtypes`.

@author:
renamed to infer_dtypes.