[SPARK-50291][PYTHON] Standardize verifySchema parameter of createDataFrame in Spark Classic

[xinrong-meng]

What changes were proposed in this pull request?

The PR targets at Spark Classic only. Spark Connect will be handled in a follow-up PR.

verifySchema parameter of createDataFrame decides whether to verify data types of every row against schema.

Now it only takes effect for with createDataFrame with

• regular Python instances

The PR proposes to make it work with createDataFrame with

• pyarrow.Table
• pandas.DataFrame with Arrow optimization
• pandas.DataFrame without Arrow optimization

By default, verifySchema parameter is pyspark._NoValue, if not provided, createDataFrame with

• pyarrow.Table, verifySchema = False
• pandas.DataFrame with Arrow optimization, verifySchema = spark.sql.execution.pandas.convertToArrowArraySafely
• pandas.DataFrame without Arrow optimization, verifySchema = True
• regular Python instances, verifySchema = True (existing behavior)

Why are the changes needed?

The change makes schema validation consistent across all formats, improving data integrity and helping prevent errors.
It also enhances flexibility by allowing users to choose schema verification regardless of the input type.

Part of SPARK-50146.

Does this PR introduce any user-facing change?

Setup:

>>> import pyarrow as pa
>>> import pandas as pd
>>> from pyspark.sql.types import *
>>> 
>>> data = {
...     "id": [1, 2, 3],
...     "value": [100000000000, 200000000000, 300000000000]
... }
>>> schema = StructType([StructField("id", IntegerType(), True), StructField("value", IntegerType(), True)])

Usage - createDataFrame with pyarrow.Table

>>> table = pa.table(data)
>>> spark.createDataFrame(table, schema=schema).show()  # verifySchema defaults to False
+---+-----------+
| id|      value|
+---+-----------+
|  1| 1215752192|
|  2|-1863462912|
|  3| -647710720|
+---+-----------+

>>> spark.createDataFrame(table, schema=schema, verifySchema=True).show()
...
pyarrow.lib.ArrowInvalid: Integer value 100000000000 not in range: -2147483648 to 2147483647

Usage - createDataFrame with pandas.DataFrame without Arrow optimization

>>> pdf = pd.DataFrame(data)
>>> spark.createDataFrame(pdf, schema=schema).show()  # verifySchema defaults to True
...
pyspark.errors.exceptions.base.PySparkValueError: [VALUE_OUT_OF_BOUNDS] Value for `obj` must be between -2147483648 and 2147483647 (inclusive), got 100000000000
>>> spark.createDataFrame(table, schema=schema, verifySchema=False).show()
+---+-----------+
| id|      value|
+---+-----------+
|  1| 1215752192|
|  2|-1863462912|
|  3| -647710720|
+---+-----------+

Usage - createDataFrame with pandas.DataFrame with Arrow optimization

>>> pdf = pd.DataFrame(data)
>>> spark.conf.set("spark.sql.execution.arrow.pyspark.enabled", True)
>>> spark.conf.get("spark.sql.execution.pandas.convertToArrowArraySafely")
'false'
>>> spark.createDataFrame(pdf, schema=schema).show()  # verifySchema defaults to "spark.sql.execution.pandas.convertToArrowArraySafely"
+---+-----------+
| id|      value|
+---+-----------+
|  1| 1215752192|
|  2|-1863462912|
|  3| -647710720|
+---+-----------+

>>> spark.conf.set("spark.sql.execution.pandas.convertToArrowArraySafely", True)
>>> spark.createDataFrame(pdf, schema=schema).show()
...
pyspark.errors.exceptions.base.PySparkValueError: [VALUE_OUT_OF_BOUNDS] Value for `obj` must be between -2147483648 and 2147483647 (inclusive), got 100000000000

>>> spark.createDataFrame(table, schema=schema, verifySchema=True).show()
...
pyarrow.lib.ArrowInvalid: Integer value 100000000000 not in range: -2147483648 to 2147483647

How was this patch tested?

Unit tests.

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

No.

[xinrong-meng]

https://issues.apache.org/jira/browse/SPARK-50146

[HyukjinKwon]

I think we don't need to mention this.

[HyukjinKwon]

Maybe say that this parameter is now respected in Spark Connect and with Arrow optimization

[xinrong-meng]

Makes sense! Removed.

[xinrong-meng]

Type hints failed weirdly as https://github.com/xinrong-meng/spark/actions/runs/11812734311/job/32908478871, ignoring #type: ignore comments.
Rebased master

[HyukjinKwon]

Merged to master.

[HyukjinKwon]

Actually had the offline discussion. I think we should evaluate the performance impact, and think about deprecating this if this isn't really useful instead of propagating it.

Let me revert #48677 and #48841 for now.