[SPARK-52215][PYTHON][CONNECT] Implement Scalar Arrow UDF - #50759

What changes were proposed in this pull request?

Implement Scalar Arrow UDF

Why are the changes needed?

Pandas UDFs (and Pandas Functions like MapInPandas) have a pandas <> arrow conversion, but:

• This conversion is not stable, and gets broken from time to time
  -- The Arrow 13 upgrade, pandas UDFs with data/time types are all broken;
  -- Weird behavior when the dataset is empty
• Pandas <> Arrow conversion is pretty expensive. Zero-copy is only possible in certain narrow cases, e.g. StringType is not supported;
• The support of complex type is not good, e.g. to support StructType series, we needs to use pd.DataFrame as a workaround;

Arrow UDF is designed to resolve above issues.

Does this PR introduce any user-facing change?

No, this PR makes the underlying implementation, under pyspark.sql.pandas.functions which is not a public module.

So the new feature is not exposed to end users for now, we will need to decide the API later.

import pyarrow as pa

from pyspark.sql import functions as sf
from pyspark.sql.pandas.functions import arrow_udf      <- `pyspark.sql.pandas.functions` is not public

df = spark.range(10).withColumn("v", sf.col("id") + 1)


@arrow_udf("long")
def multiply_arrow_func(a: pa.Array, b: pa.Array) -> pa.Array:
    assert isinstance(a, pa.Array)
    assert isinstance(b, pa.Array)
    return pa.compute.multiply(a, b)


df.select("id", "v", multiply_arrow_func("id", "v").alias("m")).show()


+---+---+---+
| id|  v|  m|
+---+---+---+
|  0|  1|  0|
|  1|  2|  2|
|  2|  3|  6|
|  3|  4| 12|
|  4|  5| 20|
|  5|  6| 30|
|  6|  7| 42|
|  7|  8| 56|
|  8|  9| 72|
|  9| 10| 90|
+---+---+---+

How was this patch tested?

new tests

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

no