Cover Spark 4.2 Arrow Python UDF columnar input [fast-ut] [reduced-it][databricks]

[firestarman]

Fixes #15663.

Description

Spark 4.2 (SPARK-56350) lets CPU ArrowEvalPythonExec consume Arrow-backed ColumnarBatch input and skip ColumnarToRow. That does not change the GPU Python UDF path (GpuColumnVector -> cuDF Arrow IPC), but it does change how tests and CPU fallbacks must behave.

• Keep the test Arrow DSv2 reader batches and allocator alive until the reader closes. Spark 4.2 CPU ArrowEvalPythonExec can retain pass-through ColumnVector references; this is CPU ownership, not GpuArrowEvalPythonExec.
• Add Arrow-source pandas UDF coverage that expects GPU HostColumnarToGpu + GpuArrowEvalPythonExec.
• Add Spark 4.2 Arrow-optimized regular UDF (evalType=101) coverage. CPU spark.sql.execution.arrow.pythonUDF.columnarInput.enabled is disabled only to avoid the Spark 4.2.0 hang from SPARK-58241; that is a CPU bug, not a GPU bug.
• Add a fallback test with a GPU Parquet scan under CPU ArrowEvalPythonExec so GpuColumnVector is not handed to Spark's CPU Arrow UDF path.

There is no user-facing config or execution-path change.

Checklists

Documentation

• Updated for new or modified user-facing features or behaviors
• No user-facing change

Testing

• Added or modified tests to cover new code paths
• Covered by existing tests
  (Please provide the names of the existing tests in the PR description.)
• Not required

New/updated tests:

• test_arrow_source_pandas_udf
• test_arrow_source_regular_udf
• test_pandas_udf_cpu_arrow_eval_after_gpu_scan

Performance

• Tests ran and results are added in the PR description
• Issue filed with a link in the PR description
• Not required

[greptile-apps]

Greptile Summary

Adds Spark 4.2 integration coverage for Arrow-backed Python UDF input and adjusts the test DataSource V2 reader’s Arrow resource lifetime.

• Keeps Arrow allocators alive while Spark consumers may retain column-vector references.
• Adds pandas and Arrow-optimized regular UDF coverage for Arrow-source batches.
• Adds fallback coverage for CPU Arrow UDF execution above a GPU Parquet scan.

Confidence Score: 5/5

The PR appears safe to merge.

No blocking failure remains.

Important Files Changed

Filename	Overview
integration_tests/src/main/python/datasourcev2_read_test.py	Adds Spark 4.2 Arrow-source pandas and regular Python UDF parity tests with explicit GPU-plan capture.
integration_tests/src/main/python/udf_test.py	Adds CPU ArrowEvalPythonExec fallback coverage above a GPU Parquet scan using bounded integer input.
integration_tests/src/main/scala/com/nvidia/spark/rapids/tests/datasourcev2/parquet/TestingV2Source.scala	Moves Arrow allocation to reader scope and defers allocator closure while returned column vectors remain retained.

Flowchart

%%{init: {'theme': 'neutral'}}%%
flowchart LR
  A[Arrow DataSource V2 batch] --> B{Execution path}
  B -->|GPU UDF| C[HostColumnarToGpu]
  C --> D[GpuArrowEvalPythonExec]
  B -->|CPU fallback| E[GPU Parquet scan]
  E --> F[Host transition]
  F --> G[ArrowEvalPythonExec]
  A --> H[Consumer releases vectors]
  H --> I[Reader allocator cleanup]

Loading

_{Reviews (3): Last reviewed commit: "Merge branch 'main' into audit/15663-arr..." | Re-trigger Greptile}

[firestarman]

build

[firestarman]

build

[res-life]

Forget to close this columnar batch list?