[SPARK-50618][SS][SQL] Make DataFrameReader and DataStreamReader leverage the analyzer more #49238
Conversation
| } | ||
| val relation = JDBCRelation(parts, options)(sparkSession) | ||
| sparkSession.baseRelationToDataFrame(relation) | ||
| Dataset.ofRows(sparkSession, UnresolvedJDBCRelation(url, table, predicates, params)) |
There was a problem hiding this comment.
shall we reuse UnresolvedDataSource with the source set as "jdbc"?
There was a problem hiding this comment.
hmm the parameters are different. I'm wondering how this is done in SQL. Seems people can't use arbitrary predicates to build JDBCPartition with SQL API.
There was a problem hiding this comment.
maybe we should add this feature to SQL first, and then unify it.
There was a problem hiding this comment.
having this available at the moment can help us unify it in SQL afterwards :) This seems to be the only edge case - seems like a new API since I looked at Spark. Do you want me to leave it as is and just migrate the file based and generic data sources?
There was a problem hiding this comment.
yea let's leave out the JDBC one for now. Once we have feature parity between SQL and DataFrame, we can revisit this.
|
Thanks for the feedback @cloud-fan! Addressed |
HyukjinKwon
changed the title
HyukjinKwon
changed the title
| assertFirstUnresolved(spark.read.format("org.apache.spark.sql.test").load()) | ||
| assertFirstUnresolved(spark.read.format("org.apache.spark.sql.test").load(dir)) | ||
| assertFirstUnresolved(spark.read.format("org.apache.spark.sql.test").load(dir, dir, dir)) | ||
| assertFirstUnresolved(spark.read.format("org.apache.spark.sql.test").load(Seq(dir, dir): _*)) | ||
| Option(dir).map(spark.read.format("org.apache.spark.sql.test").load) |
There was a problem hiding this comment.
unrelated to your PR but what does this line test?
What changes were proposed in this pull request?
Introduces two logical nodes:
UnresolvedDataSourceUnresolvedJDBCRelationThe DataFrameReader and DataStreamReader creates these unresolved nodes instead, and calls the analyzer to resolve these data sources. These then get analyzed as part of the
ResolveDataSourcerule. All logic in DataFrameReader and DataStreamReader has been moved here.There is still logic around text based format parsing on an existing Dataset. I will refactor this in a subsequent PR.
Why are the changes needed?
The DataFrameReader and DataStreamReader typically creates analyzed relations as part of their respective
.load()methods.This creates inconsistencies for what rules get applied to the query plan as part of Catalyst depending on your API of choice, such as SQL vs Python or SQL vs Scala.
The goal of this Jira is to refactor the logic in DataFrameReader and DataStreamReader classes to create unresolved plans that get analyzed as part of Catalyst.
Does this PR introduce any user-facing change?
No
How was this patch tested?
Existing unit tests and will add new tests
Was this patch authored or co-authored using generative AI tooling?
No