[WIP][SPARK-53967][PYTHON] Avoid intermediate pandas dataframe creation in df.toPandas

[zhengruifeng]

What changes were proposed in this pull request?

Avoid intermediate pandas dataframe creation in df.toPandas

before: batches -> table -> intermediate pdf -> result pdf (based on pa.Table.to_pandas)

after: batches -> table -> result pdf (based on pa.ChunkedArray.to_pandas)

Why are the changes needed?

the intermediate pandas dataframe can be skipped

simple benchmark in my local

spark.conf.set("spark.sql.execution.arrow.pyspark.enabled", "true")
spark.conf.set("spark.sql.execution.arrow.pyspark.fallback.enabled", "false")
spark.conf.set("spark.sql.execution.arrow.pyspark.selfDestruct.enabled", "true")

import time
from pyspark.sql import functions as sf


df = spark.range(1000000).select(
    (sf.col("id") % 2).alias("key"), sf.col("id").alias("v")
)
cols = {f"col_{i}": sf.lit(f"c{i}") for i in range(100)}
df = df.withColumns(cols)
df.cache()
df.count()



pdf = df.toPandas() # warm up


start_arrow = time.perf_counter()
for i in range(100):
    pdf = df.toPandas()

time.perf_counter() - start_arrow

master: 304.49954012501985 secs
this PR: 285.2997682078276 secs

Does this PR introduce any user-facing change?

no

How was this patch tested?

existing tests

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

no

[zhengruifeng]

cc @Yicong-Huang