[SPARK-54183][PYTHON][CONNECT] Avoid one intermediate temp data frame during spark connect toPandas()

### What changes were proposed in this pull request?

This PR optimizes the `to_pandas()` method in Spark Connect client to avoid creating an intermediate pandas DataFrame during Arrow to pandas conversion.

**Key changes:**
- Convert Arrow columns directly to pandas Series using `arrow_col.to_pandas()` instead of converting the entire table first with `table.to_pandas()`
- Eliminate temporary column renaming (`col_0`, `col_1`, etc.) since we no longer create an intermediate DataFrame
- Apply Spark-specific type converters directly to each Series without going through an intermediate DataFrame

### Why are the changes needed?

This optimization brings Spark Connect's `to_pandas()` implementation in line with the regular Spark DataFrame optimization made in PR #52680 ([SPARK-53967](https://issues.apache.org/jira/browse/SPARK-53967)).

**Benefits:**
1. **Reduced memory usage**: Eliminates allocation of intermediate DataFrame
2. **Better performance**: Fewer data copies, better memory locality
3. **Consistency**: Makes Spark Connect code path match the optimized regular Spark DataFrame path

### Does this PR introduce _any_ user-facing change?

No. This is a pure performance optimization with no API or behavior changes.

### How was this patch tested?
**Benchmark setup** (for manual testing):
- 1M rows × 102 columns
- Mixed types: ~25 complex columns (Date, Timestamp, Struct) + ~77 simple columns (Int, Double, String)
- Batch size: 5,000 rows per batch
- Config: Arrow enabled, self-destruct enabled

```
from pyspark.sql import SparkSession
from pyspark.sql import functions as sf
import time

spark = SparkSession.builder.remote("sc://localhost:15002").getOrCreate()

spark.conf.set("spark.sql.execution.arrow.pyspark.enabled", "true")
spark.conf.set("spark.sql.execution.arrow.pyspark.selfDestruct.enabled", "true")
spark.conf.set("spark.sql.execution.arrow.maxRecordsPerBatch", "5000")  # Small batches: 5k rows (~1.5MB/batch)

# Large dataset: 1M rows with MIXED data types
df = spark.range(1000000).select(
    sf.col("id"),
    (sf.col("id") % 2).alias("key"),
    sf.col("id").alias("v")
)

# Add various column types to test conversion performance. These types need Spark-specific conversion:
df = df.withColumns({
    "date_col_1": sf.date_add(sf.to_date(sf.lit("2024-01-01")), (sf.col("id") % sf.lit(365)).cast("int")),
    "date_col_2": sf.date_add(sf.to_date(sf.lit("2023-01-01")), (sf.col("id") % sf.lit(180)).cast("int")),
    "timestamp_col": sf.current_timestamp(),
    "struct_col_1": sf.struct(sf.col("id").cast("long").alias("a"), (sf.col("id") * sf.lit(2)).cast("long").alias("b")),
    "struct_col_2": sf.struct((sf.col("id") % sf.lit(10)).cast("int").alias("x"), (sf.col("id") / sf.lit(100.0)).alias("y")),
    "array_col": sf.array(sf.lit(1), sf.lit(2), sf.lit(3)),
    "double_col_1": sf.col("id") / sf.lit(3.14),
    "double_col_2": sf.col("id") * sf.lit(1.5) + sf.lit(100),
    "int_col": (sf.col("id") % sf.lit(1000)).cast("int"),
})

# Add more mixed columns - some simple, some complex
for i in range(45):
    if i % 5 == 0:
        df = df.withColumn(f"mixed_{i}",
            sf.date_add(sf.to_date(sf.lit("2024-01-01")), (sf.col("id") % sf.lit(i + 1)).cast("int")))
    elif i % 5 == 1:
        df = df.withColumn(f"mixed_{i}",
            sf.struct(sf.lit(i).alias("idx"), (sf.col("id") % sf.lit(i + 1)).cast("long").alias("val")))
    elif i % 5 == 2:
        df = df.withColumn(f"mixed_{i}",
            sf.concat(sf.lit(f"str_{i}_"), (sf.col("id") % sf.lit(100)).cast("string")))
    else:
        df = df.withColumn(f"mixed_{i}", (sf.col("id") * sf.lit(i) + sf.lit(i)) % sf.lit(1000))

# Add some constant strings for variety
for i in range(45):
    df = df.withColumn(f"const_{i}", sf.lit(f"c{i}"))

df = df.drop("id")
df.cache()
df.count()

# Warm up
pdf = df.toPandas()
del pdf

# Benchmark
start = time.perf_counter()
total_rows = 0
total_sum = 0

for i in range(20):
    # Convert to pandas
    pdf = df.toPandas()
    total_rows += len(pdf)
    total_sum += pdf['v'].sum()
    del pdf

    if (i + 1) % 5 == 0:
        elapsed = time.perf_counter() - start
        print(f"  {i + 1}/20 completed ({elapsed:.1f}s elapsed, ~{elapsed/(i+1):.2f}s per iteration)")

elapsed = time.perf_counter() - start
```

**Manual benchmarking results**: 6.5% improvement with mixed data types (dates, timestamps, structs, arrays, and simple types)
   - Before: 129.3s for 20 iterations (6.46s per iteration)
   - After: 120.9s for 20 iterations (6.04s per iteration)

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

Yes. Co-Genreated-by Cursor

Closes #52979 from Yicong-Huang/SPARK-54183/refactor/avoid-intermediate-df-in-topandas-connect.

Authored-by: Yicong-Huang <17627829+Yicong-Huang@users.noreply.github.com>
Signed-off-by: Ruifeng Zheng <ruifengz@apache.org>

Author: Yicong-Huang

python/pyspark/sql/connect/client/core.py

@@ -1001,52 +1001,71 @@ def to_pandas(
         schema = schema or from_arrow_schema(table.schema, prefer_timestamp_ntz=True)
         assert schema is not None and isinstance(schema, StructType)
 
-        # SPARK-51112: If the table is empty, we avoid using pyarrow to_pandas to create the
-        # DataFrame, as it may fail with a segmentation fault. Instead, we create an empty pandas
-        # DataFrame manually with the correct schema.
-        if table.num_rows == 0:
-            pdf = pd.DataFrame(columns=schema.names, index=range(0))
-        else:
-            # Rename columns to avoid duplicated column names.
-            renamed_table = table.rename_columns([f"col_{i}" for i in range(table.num_columns)])
-
-            pandas_options = {"coerce_temporal_nanoseconds": True}
-            if self_destruct == "true":
-                # Configure PyArrow to use as little memory as possible:
-                # self_destruct - free columns as they are converted
-                # split_blocks - create a separate Pandas block for each column
-                # use_threads - convert one column at a time
-                pandas_options.update(
-                    {
-                        "self_destruct": True,
-                        "split_blocks": True,
-                        "use_threads": False,
-                    }
-                )
-            pdf = renamed_table.to_pandas(**pandas_options)
-            pdf.columns = schema.names
+        # Rename columns to avoid duplicated column names during processing
+        temp_col_names = [f"col_{i}" for i in range(len(schema.names))]
+        table = table.rename_columns(temp_col_names)
+
+        # Pandas DataFrame created from PyArrow uses datetime64[ns] for date type
+        # values, but we should use datetime.date to match the behavior with when
+        # Arrow optimization is disabled.
+        pandas_options = {"coerce_temporal_nanoseconds": True}
+        if self_destruct == "true" and table.num_rows > 0:
+            # Configure PyArrow to use as little memory as possible:
+            # self_destruct - free columns as they are converted
+            # split_blocks - create a separate Pandas block for each column
+            # use_threads - convert one column at a time
+            pandas_options.update(
+                {
+                    "self_destruct": True,
+                    "split_blocks": True,
+                    "use_threads": False,
+                }
+            )
 
-        if len(pdf.columns) > 0:
+        if len(schema.names) > 0:
             error_on_duplicated_field_names: bool = False
             if struct_in_pandas == "legacy" and any(
                 _has_type(f.dataType, StructType) for f in schema.fields
             ):
                 error_on_duplicated_field_names = True
                 struct_in_pandas = "dict"
 
-            pdf = pd.concat(
-                [
-                    _create_converter_to_pandas(
-                        field.dataType,
-                        field.nullable,
-                        timezone=timezone,
-                        struct_in_pandas=struct_in_pandas,
-                        error_on_duplicated_field_names=error_on_duplicated_field_names,
-                    )(pser)
-                    for (_, pser), field, pa_field in zip(pdf.items(), schema.fields, table.schema)
-                ],
-                axis="columns",
-            )
+            # SPARK-51112: If the table is empty, we avoid using pyarrow to_pandas to create the
+            # DataFrame, as it may fail with a segmentation fault.
+            if table.num_rows == 0:
+                # For empty tables, create empty Series with converters to preserve dtypes
+                pdf = pd.concat(
+                    [
+                        _create_converter_to_pandas(
+                            field.dataType,
+                            field.nullable,
+                            timezone=timezone,
+                            struct_in_pandas=struct_in_pandas,
+                            error_on_duplicated_field_names=error_on_duplicated_field_names,
+                        )(pd.Series([], name=temp_col_names[i], dtype="object"))
+                        for i, field in enumerate(schema.fields)
+                    ],
+                    axis="columns",
+                )
+            else:
+                pdf = pd.concat(
+                    [
+                        _create_converter_to_pandas(
+                            field.dataType,
+                            field.nullable,
+                            timezone=timezone,
+                            struct_in_pandas=struct_in_pandas,
+                            error_on_duplicated_field_names=error_on_duplicated_field_names,
+                        )(arrow_col.to_pandas(**pandas_options))
+                        for arrow_col, field in zip(table.columns, schema.fields)
+                    ],
+                    axis="columns",
+                )
+            # Restore original column names (including duplicates)
+            pdf.columns = schema.names
+        else:
+            # empty columns
+            pdf = table.to_pandas(**pandas_options)
 
         if len(metrics) > 0:
             pdf.attrs["metrics"] = metrics