[SPARK-45917][PYTHON][SQL] Automatic registration of Python Data Source on startup

[HyukjinKwon]

What changes were proposed in this pull request?

This PR proposes to add the support of automatic Python Data Source registration.

End user perspective:

# Assume that `customsource` defined a short name as `custom` 
pip install pyspark_customsource

Users can directly use the Python Data Source

df = spark.format("custom").load()

Developer perspective:

The packages should follow the structure below:

• The package name should start with pyspark_ prefix
• pyspark_*.DefaultSource has to be defined that inherits pyspark.sql.datasource.DataSource

For example:

pyspark_customsource
├── __init__.py
 ...

__init__.py:

from pyspark.sql.datasource import DataSource

class DefaultSource(Datasource):
    pass

Why are the changes needed?

This allows the developers to release and maintain their 3rd party Python Data Sources separately (e.g., in PyPI), and end users can easily install the Python Data Source without doing anything other than just pip install pyspark_their_source

Does this PR introduce any user-facing change?

Yes, this allows users to pip install pyspark_custom_source, and automatically register it as Data Source available in Spark.

How was this patch tested?

Unittests were added.

Also manual test as below:

rm -fr pyspark_mysource
mkdir pyspark_mysource
cd pyspark_mysource
echo '
from pyspark.sql.datasource import DataSource, DataSourceReader, InputPartition

class TestDataSourceReader(DataSourceReader):
    def __init__(self, options):
        self.options = options
    def partitions(self):
        return [InputPartition(i) for i in range(3)]
    def read(self, partition):
        yield partition.value, str(partition.value)


class DefaultSource(DataSource):
    @classmethod
    def name(cls):
        return "test"
    def schema(self):
        return "x INT, y STRING"
    def reader(self, schema) -> "DataSourceReader":
        return TestDataSourceReader(self.options)
    @classmethod
    def name(cls):
        return "mysource"
' > __init__.py
cd ..
./bin/pyspark

spark.read.format("mysource").load().show()

+---+---+
|  x|  y|
+---+---+
|  0|  0|
|  1|  1|
|  2|  2|
+---+---+

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

No.

[dongjoon-hyun]

+1, LGTM.

[HyukjinKwon]

Let me actually add the test cases here together while I am here.

[ueshin]

Do we need Py4J path? The Python functions are not supposed to use Py4J?

[HyukjinKwon]

We don't use it but we need it when we import .. e.g., https://github.com/apache/spark/blob/master/python/pyspark/__init__.py#L53 -> https://github.com/apache/spark/blob/master/python/pyspark/conf.py#L23 when we import Python Data Source.

[HyukjinKwon]

Merged to master.

[panbingkun]

@HyukjinKwon
I have reverted #44504 (CommitID: 229a4eaf547e5c263c749bd53f7f9a89f4a9bea9).
Based on the current running results, the Run Spark on Kubernetes Integration test failure of GA is related to this.

https://github.com/apache/spark/pull/44530/files
https://github.com/panbingkun/spark/actions/runs/7353125339/job/20018716583

[HyukjinKwon]

Thx, let me fix up together at #44519