[SPARK-56324] Introducing message-based communication to Spark -> PySpark communication channel

[sven-weber-db]

What changes were proposed in this pull request?

This is the second in a series of PRs that introduce message-based communication to PySpark UDFs. This initiative is part of SPIP SPARK-55278, which proposes language-agnostic UDFs. This PR builds on top of the changes from PR #55515.

The goal of introducing message-based communication to PySpark is to:

1. Make the communication between Spark <-> PySpark more structured.
2. Enable new communication protocols (e.g., gRPC) transparently.

The overall goal is to introduce a second communication channel while keeping the existing channel intact. Specifically, we want to introduce gRPC in addition to UDS. The existing UDS channel will not be changed, and its characteristics, including performance, will remain untouched.

This PR specifically propose the following changes:

1. PythonRunner.scala - Add a new message header and a length field to the initialization data/message send from Spark to PySpark. This change is required to distinguish the initial message from other, later, messages. It constitutes the only required change in the Spark -> PySpark wire protocol.
2. Add new abstractions to read Spark -> PySpark messages from the existing socket channel - including the new init message
3. Change worker.py to use the new socket message reader to process the UDF request

With these changes, a new message reader can be implemented and transparently use for other transport channels (e.g. gRPC).

Why are the changes needed?

The changes introduced here make PySpark transport layer agnostic for the Spark -> PySpark channel. This is required for PySpark to support the new, language agnostic UDF protocol proposed in SPIP SPARK-55278.

Does this PR introduce any user-facing change?

No

How was this patch tested?

Existing test suites:

TODO:

PySpark

pyspark.tests.test_worker
pyspark.sql.tests.test_udf
pyspark.sql.tests.test_udtf

Spark

org.apache.spark.sql.execution.python.PythonUDFSuite

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