test: add seeded fuzz coverage for PyArrow UDFs

[LinSimon-901101]

Which issue does this PR close?

Closes #4384.

Rationale for this change

Add reproducible randomized coverage for the Arrow IPC path between Comet
and Python, exercising combinations of nested types, null densities, and
multiple batches alongside the existing targeted tests.

What changes are included in this PR?

• Add seeded generators for primitive and bounded nested array, struct,
  and map types, including variable-length values and decimal types.
• Generate primitive-element arrays with lengths up to 128, while keeping
  outer containers small to bound nested data growth.
• Exercise randomized per-column null densities and explicit null fractions
  of 0, 0.01, 0.5, 0.99, and 1.
• Compare accelerated and fallback mapInArrow results against a vanilla
  Spark Parquet baseline that is also checked against the generated input.
• Assert execution plans and verify actual Python input batch indices
  and sizes, including partial final batches.
• Handle the PySpark 4.2 Arrow schema conversion API change.
• Add the module to the PyArrow UDF workflow and CI change detection,
  retaining the existing tests.

How are these changes tested?

Local testing on macOS with JDK 21, Python 3.11.15, PyArrow 25.0.1,
pandas 3.0.6, and pytest 9.1.1.

The complete fuzz module passed after both the Spark 4.2 API compatibility
fix and the larger-array coverage change:

• Spark 4.0.4: 39 passed.
• Spark 4.1.3: 39 passed.
• Spark 4.2.0: 39 passed, with pandas 3.x compatibility warnings.

The end-to-end cases cover three seeds, two batch sizes, and six null
configurations, checking both accelerated and fallback execution.
Confirmed that the fixed seeds generate length-128 arrays.

Additional validation:

• python dev/ci/check-ci-config.py: passed after rebasing onto main.
• Verified that changes to the new test select the PyArrow UDF job for
  merge queue events, but not ordinary unlabeled PR updates.

[sunchao]

Correctness

Reviewed bc45ff995a7504ae138a8b73eb7ee054af7dda52 against 5ca149928f7743bfe7a96feadea5e0f9bed1412f. The existing PyArrow tests exercise targeted shapes. This adds deterministic combinations of primitive types, nested arrays/structs/maps, null densities and batch boundaries. I found no verified P1/P2 issue in the three-file change.

The baseline is independent of the accelerated path: each case writes and reads Parquet with Comet disabled and compares the Spark result with the generated rows. Both UDF modes then compare every original field with that baseline. Sorting by the complete row-ID sequence retains multiplicity, so lost and duplicated rows cannot cancel out. The plan checks require CometMapInBatch only in accelerated mode, and the worker returns batch indices and sizes that check both full batches and the partial final batch.

The generated values respect their declared types. Integer boundaries stay in range, float values are exactly representable in float32, decimal tuple construction preserves precision 38, and map keys are unique and non-null. Recursive nulls and empty collections are covered. The standalone Arrow IPC test also checks the restored schema and validates the arrays. I compared the relevant mapping, serializer and batching behavior with maintained Spark 4.0 source. UTC timestamp transport is the intended scope here. Invalid-UDF exception comparison remains in the existing targeted suite, whose invocation is retained. This does not establish coverage of every schema-metadata variation or non-UTC behavior.

Validation

python3 -B dev/ci/check-ci-config.py passed. Focused probes of the actual helper code, using lightweight type/Row doubles, verified repeatability across all 18 seed/null combinations and rejected six row corruptions plus eight batch-metadata corruptions. These were Python component checks, not PyArrow, Spark, JNI or native runs. I did not build Comet or run the end-to-end module locally.

The author reports 39 passing tests on each of Spark 4.0.4, 4.1.3 and 4.2.0. I have not independently reproduced those runs. Current Comet CI, CodeQL and title checks require approval and have zero jobs. The successful label job checked out base 5ca149928f77 and supplies no product-test evidence. Maintained Spark 3.4/4.1 sources are unavailable. Spark 3.5 has no accelerated runner and is outside this workflow's 4.x matrix.

Performance

The added cost is test execution. Each case uses 37 rows, one Parquet partition and nesting depth at most three. Only arrays with primitive elements can reach length 128, while outer containers remain small. The generator has a fixed number of iterations and bounded variable-length values. One session serves the module, and each case shares its Parquet input and baseline across both modes.

The source defines 36 end-to-end cases per Spark version, each exercising two modes, plus three Arrow IPC checks. CI runs the new module only when the existing PyArrow gate is selected. I verified selection for merge-queue and explicitly labeled PR events, and exclusion from ordinary unlabeled PR updates. No performance improvement is claimed or measured. There is no test-specific timeout in this change, so it is bounded input coverage rather than a test of cancellation or hung-worker recovery.

Design

The approach extends the existing Python-worker test harness with a local seeded generator and an independent Spark oracle. Carrying batch metadata through the UDF makes the intended multi-batch behavior observable. The four per-case configuration settings are restored in finally, and the session fixture stops Spark after the module. The schema-conversion compatibility check is confined to the Arrow IPC test.

The existing runner and buffer-ownership implementation is unchanged. The new UDF reuses input Arrow columns while constructing the output batch, consistent with the existing synchronous Arrow serialization path. I found no new ownership transition or resource-lifetime defect introduced by these tests. Keeping this focused transport coverage alongside the existing empty-input, exception and dictionary tests is a reasonable division.

Abstraction & complexity

The helpers each have a clear role: choose types, construct legal values, generate a case, normalize collected values and compare rows. A per-case random.Random(seed) keeps generation independent of test order. The seed, schema, null fraction, batch size and execution mode appear in failure context, which makes reproducing a failing case practical.

The implementation fits in one test module and two explicit CI registrations. It adds no general fuzzing framework or production abstraction. I found no actionable simplification needed before merge.