anew opened a new pull request, #59796:
URL: https://github.com/apache/spark/pull/59796
### What changes were proposed in this pull request?
Pipelines flow analysis (`FlowAnalysis.analyze`) records a flow's
dependencies on other datasets
in the pipeline by rewriting the `UnresolvedRelation`s in the flow's plan.
Some Spark Connect
operators, however, analyze their child eagerly when the flow is defined,
because they need its
schema: `withColumn`, `withColumnRenamed`, `na.fill` / `na.drop` /
`na.replace`, `to`,
`withWatermark`, and others. Once the upstream dataset exists in the
catalog, which is the case on
every run of a pipeline after the first, that eager analysis succeeds and
the read reaches flow
analysis already resolved to the catalog table. No dependency is recorded,
so the flow is
scheduled concurrently with its upstream instead of after it.
This PR makes flow analysis handle such pre-resolved reads like the
`UnresolvedRelation` they were
resolved from:
- A new extractor, `ResolvedTableRead`, recovers the `UnresolvedRelation`
(name, read options, and
whether the read is streaming) from a resolved catalog table read:
`DataSourceV2Relation`,
`StreamingRelationV2`, `LogicalRelation` over a file source,
`StreamingRelation`, and
`HiveTableRelation`. Reads it cannot recover faithfully are left
unchanged: path-based reads,
time travel reads, other V1 relations, and identifiers with more than
three parts.
- The recovered relation is read through the existing path, so a dataset in
the pipeline goes
through `readGraphInput` (recording the dependency) and an external table
is read fresh through
`readExternalBatchInput` / `readExternalStreamInput`.
- The operators above the read are already analyzed and reference the old
relation's output
attributes, so the new read is wrapped in a `Project` that exposes it
under the same attribute
IDs and types. Columns are matched by name and type, ignoring nullability
(e.g. a dataset's
inferred schema versus its table's), and cast to the original type when
they differ.
- The plan is now traversed bottom-up (`transformUpWithSubqueries`), so the
plans substituted for
reads are never traversed and matched again. This does not change how
`UnresolvedRelation`s are
handled, since they are leaves and their replacements are fully analyzed.
Reported on the dev list:
https://lists.apache.org/thread/or2hs36558lnkq971mdy5nj0vltkzto2
### Why are the changes needed?
Without the dependency, a flow can read its upstream dataset while it is
being refreshed. For
example, with
```python
@dp.materialized_view
def a():
return spark.sql("SELECT * FROM VALUES (1, 'north'), (2, 'south') AS
t(id, name)")
@dp.materialized_view
def b():
return spark.read.table("a").withColumn("m", F.upper(F.col("name")))
```
the first run is correct, but on the second run `b` starts at the same time
as `a`, reads the
table while `a` refreshes it, and writes no rows. The run still completes
successfully, and a dry
run does not detect it. With `select` instead of `withColumn` the dependency
is kept, because
`select` does not analyze its child eagerly.
Recording the dependency alone is not enough: the pre-resolved relation also
holds the file
listing of the table from when the flow was defined. When the flow is
executed, it is re-analyzed
against a subgraph in which the upstream table is external, so the stale
relation would still be
read, failing with `FAILED_READ_FILE.FILE_NOT_EXIST` after the upstream was
rewritten. Reading the
recovered relation the same way as an `UnresolvedRelation` fixes both.
### Does this PR introduce _any_ user-facing change?
Yes, bug fixes:
- A flow that reads another dataset of the pipeline through an eagerly
analyzed operator now
depends on that dataset on every run, not only on the first, so it is
scheduled after it and
reads its refreshed data.
- An external table read through an eagerly analyzed operator is now read
when the flow runs
rather than as of when the flow was defined, and is reported as a used
external input, like an
external table read without such an operator.
### How was this patch tested?
New tests in `PythonPipelineSuite`. Each runs a pipeline, then defines the
graph again against the
tables the first run materialized, changes the upstream data, and checks
both the flow's
dependencies and the data written by the second run:
- a grid over `withColumn`, `withColumnRenamed`, `na.fill`, `na.drop`,
`na.replace`, and `to`,
with `select` as a control that does not analyze eagerly. Operators
without an unresolved
fallback cannot be defined before the upstream exists, so for those the
first run uses a plain
read, as when a deployed pipeline is edited;
- a streaming read through `withWatermark`;
- a column whose nested nullability differs between the dataset's inferred
schema and its table;
- an external table read through `withColumn`, which must stay external.
Without the fix, all of these fail with `Set() did not equal
Set(`spark_catalog`.`default`.`src`)`
except the two controls. With the fix, all pass.
Also ran `build/sbt pipelines/test 'connect/testOnly
org.apache.spark.sql.connect.pipelines.*'`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: 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]
