Vivek1106-04 opened a new pull request, #59100:
URL: https://github.com/apache/spark/pull/59100
### What changes were proposed in this pull request?
Spark Connect's `withColumns` and `withColumnsRenamed` try eager analysis
first and fall back to a lazily resolved plan (`UnresolvedStarWithColumns` /
`UnresolvedStarWithColumnsRenames`) when it fails (SPARK-53505).
This PR plans Declarative Pipelines flow relations without that eager
analysis:
- `SparkConnectPlanner.transformRelationWithoutEagerAnalysis` transforms a
relation with the eager analysis of `withColumns` / `withColumnsRenamed`
disabled, and `handlePipelineCommand` hands it to `PipelinesHandler` instead of
`transformRelation`. `PipelinesHandler` uses it only in `defineFlow`.
- Both call sites now go through a shared `tryEagerAnalysis` helper, which
returns the lazy plan directly when eager analysis is disabled and otherwise
behaves exactly as before.
Nothing changes outside pipelines, and no configuration is added.
### Why are the changes needed?
`FlowAnalysis` discovers a flow's dependencies by matching
`UnresolvedRelation`s in the flow's plan. `PipelinesHandler.defineFlow` runs
the flow relation through the planner at `DefineFlow` time:
- **First run (clean catalog):** the upstream table does not exist, eager
analysis fails, the fallback keeps the `UnresolvedRelation`, and the dependency
is recorded.
- **Every later run:** the upstream table now exists, eager analysis
succeeds and consumes the `UnresolvedRelation` against the *stale catalog
table*. `FlowAnalysis` records no dependency, and the downstream flow is
scheduled concurrently with its own upstream.
Result from the JIRA repro: on the second run `b =
spark.read.table("a").withColumn(...)` starts before `a` and ends up with **0
rows**, while the equivalent `select(...)` flow `c` is correct. The bug passes
every first-run smoke test and only shows up on scheduled re-runs.
### Does this PR introduce _any_ user-facing change?
Yes, a bug fix. Pipeline flows that use `withColumn(s)` /
`withColumnRenamed` / `withColumnsRenamed` on another pipeline dataset now keep
their dependency on that dataset across runs, so they wait for it instead of
reading a stale or empty table. Non-pipeline Spark Connect queries are
unaffected.
### How was this patch tested?
Added `SPARK-59663: withColumns and withColumnsRenamed keep the dependency
on an internal dataset that already exists in the catalog` to
`PythonPipelineSuite`. It pre-creates `spark_catalog.default.src` (simulating a
previous run), defines `src` plus a `withColumn` flow and a `withColumnRenamed`
flow on it, and asserts each flow's inputs are `{src}`.
- Without the fix, the test fails for both flows (each verified separately):
`Set() did not equal Set(spark_catalog.default.src)`.
- With the fix, `PythonPipelineSuite`, `SparkConnectPlannerSuite`,
`SparkConnectProtoSuite` and `PipelinesHandlerSuite` pass (222 tests).
```
build/sbt 'connect/testOnly *PythonPipelineSuite *SparkConnectPlannerSuite
*SparkConnectProtoSuite *PipelinesHandlerSuite'
```
Note: if user code explicitly analyzes a DataFrame *outside* the flow
function (e.g. `df.columns`), the resulting plan can come from the session plan
cache. That path already fails on a clean catalog, and the existing test
`reading internal datasets outside query function that trigger eager analysis
or execution will fail` covers it as unsupported, so it is left out of scope
here.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (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]
