[SPARK-58601][PYTHON] Tighten mapInPandas return-value contract to require a strict Iterator

[Yicong-Huang]

What changes were proposed in this pull request?

Tighten the mapInPandas (SQL_MAP_PANDAS_ITER_UDF) return-value contract in worker.py to require a strict Iterator, matching mapInArrow (SQL_MAP_ARROW_ITER_UDF).

Previously the runtime check was isinstance(result, Iterator) or hasattr(result, "__iter__"), which accepted any iterable (e.g. a returned list). It is now isinstance(result, Iterator); a non-Iterator iterable is rejected with UDF_RETURN_TYPE ("iterator of pandas.DataFrame"/"pandas.Series"). The comment explaining why verify_return_type was not reused is removed accordingly.

Why are the changes needed?

The declared signature has always been PandasMapIterFunction = Callable[[Iterator[DataFrameLike]], Iterator[DataFrameLike]], and the DataFrame.mapInPandas docstring states the function "outputs an iterator of pandas.DataFrames". The runtime, however, leniently accepted any iterable, diverging from the documented contract. This aligns the runtime with the declared Iterator[...] signature, mirroring the same tightening already done for mapInArrow in SPARK-56612. It also unblocks reusing verify_return_type at the mapInPandas site with Iterator[...] (SPARK-58598).

Does this PR introduce any user-facing change?

No. Minor tightening: a UDF returning a non-Iterator iterable (e.g. list) is now rejected with UDF_RETURN_TYPE, aligning the runtime with the documented Iterator[...] signature. This mirrors the mapInArrow tightening in SPARK-56612.

How was this patch tested?

Updated test_pandas_map.py: removed the "returning list of DataFrames" positive case from test_map_in_pandas, and added a list_not_iter negative case to check_other_than_dataframe_iter asserting a returned list is rejected, mirroring test_arrow_map.py::test_other_than_recordbatch_iter. Ran pyspark.sql.tests.pandas.test_pandas_map and the Connect parity suite.

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

No

[uros-b]

Should we update the migration guide? To explain mapInPandas UDF returning a plain list of DataFrames (e.g. lambda it: [pdf for pdf in it]) previously succeeded and now raises UDF_RETURN_TYPE with "but is list"

Otherwise looks good, thank you @Yicong-Huang!

[Yicong-Huang]

Should we update the migration guide? To explain mapInPandas UDF returning a plain list of DataFrames (e.g. lambda it: [pdf for pdf in it]) previously succeeded and now raises UDF_RETURN_TYPE with "but is list"

Thanks @uros-b! I am on the fence here. The public doc declares Callable[[Iterator[DataFrameLike]], and accepting list which is an iterable not iterator has always been a silently allowed side effect. This PR tightens the allowed return type to its declared type, so not sure how to mention it in migration guide. Maybe @HyukjinKwon @cloud-fan can suggest action here?

Actually I think we should just keep the change for now, and maybe issue a warning or sth

[cloud-fan]

0 blocking, 1 non-blocking, 0 nits.
The implementation and focused regression test are sound, but the observable compatibility change should be called out in the PySpark migration guide.

Suggestions (1)

• General: Add a PySpark 4.3 migration note explaining that mapInPandas now rejects lists and other non-Iterator iterables, and recommend returning iter(...).

Verification

Verified that the public mapInPandas documentation requires an iterator return, the worker now performs a strict collections.abc.Iterator check, and the closest mapInArrow analogue enforces the same outer-container contract through verify_return_type. The new test covers the formerly accepted list case and preserves existing coverage for invalid iterator elements.

PR metadata suggestions

• Correct the user-facing-change section: returning a list or another non-Iterator iterable now changes from success to UDF_RETURN_TYPE.