fix + test

Author: xinrong-meng

python/pyspark/pandas/data_type_ops/base.py

@@ -417,16 +417,14 @@ def ge(self, left: IndexOpsLike, right: Any) -> SeriesOrIndex:
         raise TypeError(">= can not be applied to %s." % self.pretty_name)
 
     def eq(self, left: IndexOpsLike, right: Any) -> SeriesOrIndex:
-        from pyspark.pandas.internal import InternalField
-
         if is_ansi_mode_enabled(left._internal.spark_frame.sparkSession):
             if _should_return_all_false(left, right):
-                return left._with_new_scol(F.lit(False))
+                return left._with_new_scol(F.lit(False)).rename(None)
 
         if isinstance(right, (list, tuple)):
             from pyspark.pandas.series import first_series, scol_for
             from pyspark.pandas.frame import DataFrame
-            from pyspark.pandas.internal import NATURAL_ORDER_COLUMN_NAME
+            from pyspark.pandas.internal import NATURAL_ORDER_COLUMN_NAME, InternalField
 
             if len(left) != len(right):
                 raise ValueError("Lengths must be equal")

python/pyspark/pandas/data_type_ops/num_ops.py

@@ -184,7 +184,7 @@ def eq(self, left: IndexOpsLike, right: Any) -> SeriesOrIndex:
         else:
             if is_ansi_mode_enabled(left._internal.spark_frame.sparkSession):
                 if _should_return_all_false(left, right):
-                    return left._with_new_scol(F.lit(False))
+                    return left._with_new_scol(F.lit(False)).rename(None)
                 if _is_boolean_type(right):
                     right = transform_boolean_operand_to_numeric(
                         right, spark_type=as_spark_type(left.dtype)

python/pyspark/pandas/tests/data_type_ops/test_num_ops.py

@@ -23,6 +23,7 @@
 from pyspark import pandas as ps
 from pyspark.pandas.config import option_context
 from pyspark.testing.pandasutils import PandasOnSparkTestCase
+from pyspark.testing.utils import is_ansi_mode_test
 from pyspark.pandas.tests.data_type_ops.testing_utils import OpsTestBase
 from pyspark.pandas.typedef.typehints import (
     extension_dtypes_available,
@@ -128,6 +129,18 @@ def test_invert(self):
             else:
                 self.assertRaises(TypeError, lambda: ~psser)
 
+    def test_comparison_dtype_compatibility(self):
+        pdf = pd.DataFrame(
+            {"int": [1, 2], "bool": [True, False], "float": [0.1, 0.2], "str": ["1", "2"]}
+        )
+        psdf = ps.from_pandas(pdf)
+        self.assert_eq(pdf["int"] == pdf["bool"], psdf["int"] == psdf["bool"])
+        self.assert_eq(pdf["bool"] == pdf["int"], psdf["bool"] == psdf["int"])
+        self.assert_eq(pdf["int"] == pdf["float"], psdf["int"] == psdf["float"])
+        if is_ansi_mode_test:  # TODO: match non-ansi behavior with pandas
+            self.assert_eq(pdf["int"] == pdf["str"], psdf["int"] == psdf["str"])
+        self.assert_eq(pdf["float"] == pdf["bool"], psdf["float"] == psdf["bool"])
+
     def test_eq(self):
         pdf, psdf = self.pdf, self.psdf
         for col in self.numeric_df_cols: