[SPARK-52593][PS] Avoid CAST_INVALID_INPUT of Series.dot and DataFrame.dot in ANSI mode

[xinrong-meng]

What changes were proposed in this pull request?

Avoid CAST_INVALID_INPUT of Series.dot and DataFrame.dot in ANSI mode

Why are the changes needed?

Ensure pandas on Spark works well with ANSI mode on.
Part of https://issues.apache.org/jira/browse/SPARK-52556.

Does this PR introduce any user-facing change?

Yes. Series.dot raises expected error in ANSI, for example

>>> pdf = pd.DataFrame({"a": [1, 2, 3], "b": [4, 5, 6]})
>>> psdf = ps.from_pandas(pdf)
>>> psdf_other = ps.DataFrame({"a": [1, 2, 3], "b": [4, 5, 6]}, index=["x", "y", "z"])
>>> ps.set_option("compute.ops_on_diff_frames", True)

FROM

>>> psdf["b"].dot(psdf_other)
...
25/06/27 15:08:57 ERROR Executor: Exception in task 0.0 in stage 14.0 (TID 98)
org.apache.spark.SparkNumberFormatException: [CAST_INVALID_INPUT] The value 'x' of the type "STRING" cannot be cast to "BIGINT" because it is malformed. Correct the value as per the syntax, or change its target type. Use `try_cast` to tolerate malformed input and return NULL instead. SQLSTATE: 22018
...

TO

>>> psdf["b"].dot(psdf_other)
Traceback (most recent call last):
...
ValueError: matrices are not aligned
>>> 

SAME AS ANSI OFF

>>> spark.conf.set("spark.sql.ansi.enabled", False)
>>> psdf["b"].dot(psdf_other)
Traceback (most recent call last):
...
ValueError: matrices are not aligned

How was this patch tested?

Unit tests

(dev3.11) spark (series_dot) % SPARK_ANSI_SQL_MODE=true  ./python/run-tests --python-executables=python3.11 --testnames "pyspark.pandas.tests.series.test_series SeriesTests.test_dot"
...
Finished test(python3.11): pyspark.pandas.tests.series.test_series SeriesTests.test_dot (6s)
Tests passed in 6 seconds

Was this patch authored or co-authored using generative AI tooling?

No

[xinrong-meng]

@ueshin may I get a review please?

[ueshin]

I guess this means index.sort_values() has CAST_INVALID_INPUT issue?

[xinrong-meng]

It's equals

>>> psidx1 = ps.Index(['x', 'y', 'z'])                     
>>> psidx2 = ps.Index([1, 2, 3])
>>> psidx1.sort_values().equals(psidx2.sort_values())
Traceback (most recent call last):
...
pyspark.errors.exceptions.captured.NumberFormatException: [CAST_INVALID_INPUT] The value 'x' of the type "STRING" cannot be cast to "BIGINT" because it is malformed. Correct the value as per the syntax, or change its target type. Use `try_cast` to tolerate malformed input and return NULL instead. SQLSTATE: 22018
== DataFrame ==
"__eq__" was called from
<stdin>:1

[xinrong-meng]

I know this collects the index to the driver which is not ideal, but I haven’t found an alternative yet

[ueshin]

Let's fix equals, then. We should fix it anyway.

[ueshin]

The other comparisons, as well? like !=, <, <=, >, and >=?