[SPARK-55583][PYTHON] Validate Arrow schema types in Python data source

[Yicong-Huang]

What changes were proposed in this pull request?

This PR adds Arrow schema type validation for the pa.RecordBatch code path in Python data source reads. The fix adds a pa_schema.equals(first_element.schema) check after the existing column name validation in records_to_arrow_batches(), raising a clear DATA_SOURCE_RETURN_SCHEMA_MISMATCH error with the expected and actual Arrow schemas.

Why are the changes needed?

When a Python data source returns a pa.RecordBatch with data types that don't match the declared schema, the resulting JVM-side errors are confusing and do not indicate the root cause. For example:

• IllegalArgumentException: not all nodes, buffers and variadicBufferCounts were consumed from VectorLoader.load()
• UnsupportedOperationException: Cannot call the method "getUTF8String" of ArrowColumnVector$ArrowVectorAccessor

These errors give no indication that the issue is a schema type mismatch in the Python data source's read() method.

Does this PR introduce any user-facing change?

Yes. Previously, returning a pa.RecordBatch with mismatched types from a Python data source would result in cryptic JVM errors. Now it raises a clear DATA_SOURCE_RETURN_SCHEMA_MISMATCH error showing the expected and actual Arrow schemas.

How was this patch tested?

Added a test case in test_python_datasource.py::test_arrow_batch_data_source.

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

No

[Yicong-Huang]

cc @allisonwang-db

[allisonwang-db]

What's the full error message here? Does it contain two error classes?

[Yicong-Huang]

the full message would look like

  [PYTHON_STREAMING_DATA_SOURCE_RUNTIME_ERROR]
  Failed when Python streaming data source perform planPartitions:
  PySparkRuntimeError: [DATA_SOURCE_RETURN_SCHEMA_MISMATCH]
  Return schema mismatch in the result from 'read' method.
  Expected: <expected_schema>, Found: <actual_schema>

Yes, the error contains two error classes. The outer PYTHON_STREAMING_DATA_SOURCE_RUNTIME_ERROR wraps the inner DATA_SOURCE_RETURN_SCHEMA_MISMATCH error. We use a general outer class to provide consistent error handling across all operations in streaming, while preserving the original Python error message for debugging.