dongjoon-hyun opened a new pull request, #59735:
URL: https://github.com/apache/spark/pull/59735
### What changes were proposed in this pull request?
When `PipelinesHandler` handles a `DefineFlow` command, it transforms the
relation of the flow with `SparkConnectPlanner.transformRelation`.
`spark.sql(query, df=...)` with DataFrame arguments sends a SQL-with-references
`WITH_RELATIONS` relation, and `transformRelation` handles it with
`executeSQLWithRefs`. That method analyzes each reference, registers it as a
temporary view, and runs the root SQL with `session.sql`, all at flow
registration.
This PR adds `SparkConnectPlanner.transformPipelineFlowRelation` and passes
it to `PipelinesHandler` instead of `transformRelation`. While it transforms
the relation of a flow, `transformSqlWithRefs` does not execute a valid
SQL-with-references `WITH_RELATIONS`. It parses the root SQL and binds each
reference to its alias as a CTE relation (`UnresolvedWith`). `FlowAnalysis`
already substitutes CTEs before it resolves `UnresolvedRelation`s as pipeline
inputs, so it handles a dataset that a DataFrame argument reads in the same way
as a dataset in the SQL text.
`handleSqlCommand` and all other Spark Connect requests still use
`executeSQLWithRefs`, so `spark.sql` with DataFrame arguments outside of a flow
definition is unchanged.
### Why are the changes needed?
Inside a flow function, `spark.sql` does not run the SQL. The server returns
the relation as is (SPARK-54452), so that the flow analysis can resolve it
later. With DataFrame arguments, however, `DefineFlow` analyzes it right away:
1. The flow fails to register when a DataFrame argument reads a dataset of
the same pipeline that does not exist in the catalog: a table or materialized
view on the first run, or a temporary view on every run.
```python
@dp.materialized_view
def upstream():
return spark.range(5)
@dp.materialized_view
def downstream():
return spark.sql("SELECT * FROM {df}",
df=spark.read.table("upstream"))
```
```
pyspark.errors.exceptions.connect.AnalysisException:
[TABLE_OR_VIEW_NOT_FOUND] The table or view `upstream` cannot be found. ...
```
2. When the table already exists, for example from a previous run, the flow
registers an analyzed plan. The flow analysis then does not see the read of
`upstream` and does not record it as an input of `downstream`. Without the
dependency, `downstream` can run before `upstream` is refreshed and read stale
data, without an error.
3. A command at the root that the SDP guard allows, such as `SHOW TABLES`,
is executed at flow registration.
The same query without DataFrame arguments, `spark.sql("SELECT * FROM
upstream")`, works because its relation is only parsed at flow registration.
In Spark 4.1.0 and 4.2.0, the SDP guard in `handleSqlCommand` also analyzed
the relation, so the error in (1) was raised at the `spark.sql` call in the
flow function. SPARK-57189 (#59425) moved it to flow registration.
### Does this PR introduce _any_ user-facing change?
Yes. In a flow function, `spark.sql` with DataFrame arguments can read
datasets of the same pipeline, and the flow analysis records them as inputs of
the flow. The SQL is analyzed during graph resolution instead of flow
registration, so an analysis error is reported as a flow resolution failure, as
for `spark.sql` without DataFrame arguments.
In a flow definition, DataFrame arguments are now bound as CTE relations
instead of temporary views. A statement that refers to a DataFrame argument
other than as a table in a query, such as `DESCRIBE TABLE {df}`, can no longer
resolve it when a flow function returns the statement.
### How was this patch tested?
Added two tests to `PythonPipelineSuite`. Both fail without this fix.
- `SPARK-60494: referencing internal datasets through DataFrame arguments of
spark.sql`: a new pipeline reads a materialized view with batch and streaming
reads, and a temporary view with a batch read, all through DataFrame arguments.
The test checks the inputs and streaming-ness of the flows, runs the pipeline,
and checks the results. Without the fix, registration of the first flow fails
with `TABLE_OR_VIEW_NOT_FOUND`.
- `SPARK-60494: referencing an internal dataset through DataFrame arguments
of spark.sql tracks the dependency when its table already exists`: Without the
fix, the inputs of the flow are empty.
Also passed the existing suites in `org.apache.spark.sql.connect.pipelines`,
including `PythonPipelineSuite` and `SparkDeclarativePipelinesServerSuite`, as
well as `SparkConnectPlannerSuite` and `SparkConnectProtoSuite`:
```
build/sbt 'connect/testOnly org.apache.spark.sql.connect.pipelines.*
*SparkConnectPlannerSuite *SparkConnectProtoSuite'
```
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
