cloud-fan commented on PR #59100:
URL: https://github.com/apache/spark/pull/59100#issuecomment-5888030866
Thanks @Vivek1106-04 for testing #59131 and for the benchmarks. You're right
about the regression. The eager analysis of each `withColumns` also resolves
the operators below it, and that's what keeps loops mixing `withColumn` with
`filter` / `select` / `join` under `spark.sql.analyzer.maxIterations`. The
fixed-point analyzer has no safe general fix for that today (that's what the
single-pass resolver is for), so I'll close #59131 and keep the eager analysis.
That brings us back to SDP. I think the fix belongs in how SDP plans a flow,
not in the eager analysis. As you noticed, the problem isn't specific to
`withColumns`. Any relation the Connect planner analyzes eagerly resolves the
table at `DefineFlow` time, before `FlowAnalysis` can substitute the pipeline
dataset. For relations without a lazy fallback it's worse: the flow can't even
be defined on the first run. On master:
| flow | first run (dataset not in the catalog yet) | re-run (dataset exists
from the previous run) |
|---|---|---|
| `spark.read.table("src").drop(...)` | `DefineFlow` fails with
`TABLE_OR_VIEW_NOT_FOUND` | defined, but inputs are `[]` |
| `spark.readStream.table("events").withWatermark(...)` | `DefineFlow` fails
with `TABLE_OR_VIEW_NOT_FOUND` | defined, but inputs are `[]` |
| `spark.read.table("src").withColumn(...)` | inputs `[src]` | inputs `[]`
(this PR's case) |
| `spark.sql("SELECT * FROM src").withColumn(...)` | inputs `[src]` | inputs
`[]` |
**Proposal: plan the flow when it's analyzed, and let the pipeline resolve
the table reads.**
- `PipelinesHandler.defineFlow` keeps the flow's proto relation instead of
planning it. The flow function plans it when `FlowAnalysis` analyzes the flow.
- While planning a flow, `SparkConnectPlanner` passes every plan that reads
a table (`spark.read.table`, `readStream.table`, and `spark.sql` text) through
`FlowAnalysis`'s dataset-read resolution first. The eager analysis then sees
the pipeline's inputs, and the dependencies get recorded. In this mode the
planner doesn't use the session's plan cache.
- `FlowAnalysis` gets a `createFlowFunction(buildPlan)` variant that hands
this read resolution to the plan builder. `createFlowFunctionFromLogicalPlan`,
used by SQL flows, keeps its behavior.
```scala
func = FlowAnalysis.createFlowFunction { resolveDatasetReads =>
new SparkConnectPlanner(sessionHolder)
.transformPipelineFlowRelation(relationFlowDetails.getRelation,
resolveDatasetReads)
}
```
Prototype:
[cloud-fan:sdp-deferred-flow-planning](https://github.com/cloud-fan/spark/tree/sdp-deferred-flow-planning)
([diff against
master](https://github.com/apache/spark/compare/master...cloud-fan:spark:sdp-deferred-flow-planning)),
about 70 lines, no tests yet. With it:
- Every case in the table gets the right inputs on both runs, including
`drop` and `withWatermark` on the first run. It should also cover the `na.*`
cases you listed, which I haven't tested.
- A 200-round `withColumn` + `filter` loop still resolves, since the eager
analysis is unchanged. I sent it as raw protos, because the Python client can't
send a plan that deep.
- The Scala pipeline suites I ran and the Connect planner suites pass. I
haven't run `PythonPipelineSuite`.
Known gap: for `spark.sql("... {df} ...", df=...)`, the planner registers
each DataFrame argument as a temporary view and analyzes the SQL text right
away with `session.sql`. The DataFrame arguments are covered, but a table named
directly in that SQL text is still bound to the catalog. Closing that gap
within this approach would mean copying the temp-view scoping into
`FlowAnalysis`'s own name resolution. I think the cleaner follow-up is to
resolve pipeline datasets in the analyzer's relation resolution, scoped to the
flow's analysis. Then CTEs, temporary views and the like follow the analyzer's
own rules, and `FlowAnalysis` no longer needs a separate pass over the plan.
On the points raised so far:
- @ueshin, about the perf concern from #52253: nothing changes for regular
Connect, and flows keep the eager analysis too, just at graph resolution time.
So I don't think we need a conf.
- Errors in a flow's relation now show up when the graph is resolved,
attributed to the flow, instead of at `DefineFlow`.
- The eager analysis now sees the flow's own SQL confs rather than the
session's, which I think is more correct.
Would you be up for taking this approach in this PR? Your SPARK-59663 test
still applies, and it'd be good to add `drop` and `withWatermark` cases for
both the first run and a re-run. Feel free to start from the prototype.
--
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]
