[SPARK-42457][CONNECT] Adding SparkSession#read

[zhenlineo]

What changes were proposed in this pull request?

Add SparkSession Read API to read data into Spark via Scala Client:

DataFrameReader.format(…).option(“key”, “value”).schema(…).load()

The following methods are skipped by the Scala Client on purpose:

[info]   deprecated method json(org.apache.spark.api.java.JavaRDD)org.apache.spark.sql.Dataset in class org.apache.spark.sql.DataFrameReader does not have a correspondent in client version
[info]   deprecated method json(org.apache.spark.rdd.RDD)org.apache.spark.sql.Dataset in class org.apache.spark.sql.DataFrameReader does not have a correspondent in client version
[info]   method json(org.apache.spark.sql.Dataset)org.apache.spark.sql.Dataset in class org.apache.spark.sql.DataFrameReader does not have a correspondent in client version
[info]   method csv(org.apache.spark.sql.Dataset)org.apache.spark.sql.Dataset in class org.apache.spark.sql.DataFrameReader does not have a correspondent in client version

Why are the changes needed?

To read data from csv etc. format.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

E2E, Golden tests.

[hvanhovell]

You can also set path/paths in the options. How does this work? Does the server reconcile all these path options?

[zhenlineo]

@hvanhovell I do not think we can anything special here. The server planner will call the original SQL code and merge it based on the config settings. e.g.

def load(paths: String*): DataFrame = {
....
    val legacyPathOptionBehavior = sparkSession.sessionState.conf.legacyPathOptionBehavior
    if (!legacyPathOptionBehavior &&
        (extraOptions.contains("path") || extraOptions.contains("paths")) && paths.nonEmpty) {
      throw QueryCompilationErrors.pathOptionNotSetCorrectlyWhenReadingError()
    }

    DataSource.lookupDataSourceV2(source, sparkSession.sessionState.conf).flatMap { provider =>
      DataSourceV2Utils.loadV2Source(sparkSession, provider, userSpecifiedSchema, extraOptions,
        source, paths: _*)
    }.getOrElse(loadV1Source(paths: _*))

If we merge them on the client we by default assumed the config value. So the best is actually leave the result to server SQL API to handle the merging of paths.

[hvanhovell]

Alright, that works for me. Can you make sure we test this?

[hvanhovell]

Looks good overall. Please provide clarity on path handling.

[hvanhovell]

@zhenlineo can you make the PR description a bit more descriptive?

[hvanhovell]

Oh and can you add the ticket to the title?

[amaliujia]

So it is ok to not have e2e tests for read API (similarly for the write side)?

[amaliujia]

Why we need the physical data files checked in? If we only compare the plans then we only need a fake data path?

[zhenlineo]

The server test ProtoToParsedPlanTestSuite runs this test with data. Yet, it does not verify the read data result correctness. (e.g. the content of the csv is loaded correctly or not)

[amaliujia]

I might be wrong but I am thinking if you offer a schema in the proto then the server side might not need to load the data: it loads the data as it needs to infer the scheme by reading the data directly when a schema is not set. I am not sure though except this case if server side still has to read the data directly.

[zhenlineo]

Umm, tried with json file, but the schema seems not enough. :(

[amaliujia]

ah sure :)

[amaliujia]

LGTM

Looks like you only need ./build/mvn -Pscala-2.12 scalafmt:format -Dscalafmt.skip=false -Dscalafmt.validateOnly=false -Dscalafmt.changedOnly=false -pl connector/connect/common -pl connector/connect/server -pl connector/connect/client/jvm to fix the style.

[hvanhovell]

LGTM