You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "cloud-fan (via GitHub)" <gi...@apache.org> on 2026/09/29 07:34:46 UTC
[PR] [SPARK-59855][SQL][CONNECT] Analyze long withColumns chains in one pass and remove the eager analysis in Spark Connect [spark]
cloud-fan opened a new pull request, #59131:
URL: https://github.com/apache/spark/pull/59131
### What changes were proposed in this pull request?
Spark Connect plans each `withColumns` / `withColumnsRenamed` call as a lazy `Project(Seq(UnresolvedStarWithColumns(...)), child)` (SPARK-50601), so a loop of `df = df.withColumn(...)` becomes a stack of such Projects. The fixed-point analyzer resolves a Project only after its child is resolved, and resolving one Project takes several rules (column resolution, function resolution, type coercion). So it resolves at most one Project of the chain per iteration of the `Resolution` batch, and a chain of about 100 calls hits `spark.sql.analyzer.maxIterations`. SPARK-53505 worked around this in the Connect planner: it analyzes each `withColumns` eagerly, and falls back to the lazy plan when that fails.
This PR fixes the analyzer and removes the workaround:
1. When `ResolveReferences` expands the star of a lazily planned `withColumns` whose child is resolved, it resolves that Project right away:
- It resolves the column references against the child, the same way it does for any Project. This covers names, Spark Connect `df["col"]` plan ids, metadata columns, outer references, and SQL variables.
- It runs only the `Resolution` batch on the expressions that are still unresolved, over a `LocalRelation` with the same output as the child. Then it puts the resolved expressions back into the Project.
The cost depends on the width of the Project, not the depth of its child. The parent Project is then resolved in the same traversal, so the whole chain is resolved in one pass.
2. Some expressions can't be resolved this way, and those Projects are left to the normal iterations, so results and errors stay the same. This covers:
- generators and window functions, which need the plan to be rewritten;
- subqueries;
- stars inside expressions;
- lateral column aliases;
- errors.
3. A new internal config, `spark.sql.analyzer.resolveWithColumnsInIsolation` (default `true`), turns this off.
4. The Spark Connect planner plans `withColumns` and `withColumnsRenamed` lazily again, without the eager analysis added by SPARK-53505.
### Why are the changes needed?
The eager analysis in the Connect planner is a workaround with its own problems:
- **It is quadratic.** Each call analyzes the plan built so far again. In a local benchmark of 1000 chained `withColumn` calls, analysis took about 50-65 s with the eager analysis, and about 7-9 s with this PR:
| 1000 chained calls | eager analysis | this PR |
|---|---|---|
| `abs(id)` | 52 s | 7.3 s |
| `abs(prev + 1)` | 64 s | 8.8 s |
| `abs(df[prev] + 1)` (plan-id reference) | n/a | 7.7 s |
- **It binds the plan when it's built, not when it's analyzed.** For example, in Spark Declarative Pipelines, take a flow `spark.read.table("src").withColumn(...)` where `src` is a dataset that already exists in the catalog from a previous run. The flow is bound to the existing table and loses its dependency on `src` (see #59100).
### Does this PR introduce _any_ user-facing change?
Yes, in Spark Connect:
- Long chains of `withColumn` / `withColumns` / `withColumnRenamed` calls are analyzed faster.
- `withColumns` and `withColumnsRenamed` no longer bind their input when the plan is built. For example, a pipeline flow that calls them on a dataset of the pipeline now keeps its dependency on that dataset.
Results and error messages are unchanged.
### How was this patch tested?
New tests:
- `AnalysisSuite`:
- A chain of 120 lazily planned `withColumns` fails with max iterations without the change. With the change, it resolves to the same plan as the normal iterations.
- `withColumns` that can't be resolved in isolation still resolve to the same plans as the normal iterations. The cases are a generator, a window function, a lateral column alias, a subquery, a star, and a mix with `withColumnsRenamed`.
- Errors are the same as with the normal iterations.
- `SparkConnectPlannerSuite`: a chain of 200 `withColumns` is planned lazily, analyzed, and returns the correct result. It references columns both by name and through `df[...]`.
Existing suites run locally:
- catalyst: `org.apache.spark.sql.catalyst.analysis.*`
- Connect: `SparkConnectPlannerSuite`, `SparkConnectProtoSuite`
- sql/core: `DataFrameSuite`, `DataFrameSubquerySuite`, `ColumnExpressionSuite`, `DataFrameWindowFunctionsSuite`, `GeneratorFunctionSuite`, `UDFSuite`, `DataFrameSelfJoinSuite`, `FileMetadataStructSuite`, `MetadataColumnSuite`, `DatasetSuite`
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 5.5)
This pull request and its description were written by Isaac.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59855][SQL][CONNECT] Analyze long withColumns chains in one pass and remove the eager analysis in Spark Connect [spark]
Posted by "sarutak (via GitHub)" <gi...@apache.org>.
sarutak commented on code in PR #59131:
URL: https://github.com/apache/spark/pull/59131#discussion_r4131158395
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -677,7 +677,18 @@ class Analyzer(
ResolveTranspiledPythonUDFOptions) ++
Seq(ResolveUpdateEventTimeWatermarkColumn) ++
extendedResolutionRules ++
- Seq(NameStreamingSources) : _*),
+ Seq(NameStreamingSources) : _*)
+
+ /**
+ * Runs only the "Resolution" batch. [[ResolveReferences]] uses it to resolve a lazily planned
+ * `withColumns` in isolation from its child.
+ */
+ private lazy val resolutionBatchExecutor = new RuleExecutor[LogicalPlan] {
Review Comment:
`resolutionBatch` is declared to return `Batch`, so the `asInstanceOf[Batch]` in `resolutionBatchExecutor.batches` looks redundant. If it is needed because of the path-dependent `Batch` type across the two `RuleExecutor` instances, a short comment would explain it; otherwise it can be dropped.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisSuite.scala:
##########
@@ -1032,6 +1032,75 @@ class AnalysisSuite extends AnalysisTest with Matchers {
}
}
+ private def withColumns(child: LogicalPlan, name: String, expr: Expression): LogicalPlan =
+ Project(Seq(UnresolvedStarWithColumns(Seq(name), Seq(expr))), child)
+
+ private def fn(name: String, args: Expression*): UnresolvedFunction =
+ UnresolvedFunction(name, args, isDistinct = false)
+
+ // Analyzes the plan with the lazily planned `withColumns` resolved in isolation, and checks
+ // that the result is the same as resolving them in the normal iterations.
+ private def checkWithColumnsResolution(plan: LogicalPlan): Unit = {
+ val expected = withSQLConf(
+ SQLConf.ANALYZER_RESOLVE_WITH_COLUMNS_IN_ISOLATION.key -> "false",
+ SQLConf.ANALYZER_MAX_ITERATIONS.key -> "1000") {
+ getAnalyzer.executeAndCheck(plan, new QueryPlanningTracker)
+ }
+ comparePlans(getAnalyzer.executeAndCheck(plan, new QueryPlanningTracker), expected)
+ }
+
+ test("SPARK-59855: resolve a long chain of lazily planned withColumns") {
+ // Like a chain of Spark Connect `withColumn` calls. Each call adds a column that references
+ // the column added by the previous call and takes several rules to be resolved.
+ val chain = (1 to 120).foldLeft[LogicalPlan](testRelation) { (child, i) =>
+ val prev = UnresolvedAttribute(if (i == 1) "a" else s"c${i - 1}")
+ withColumns(child, s"c$i", fn("abs", prev + 1L))
+ }
+ withSQLConf(SQLConf.ANALYZER_RESOLVE_WITH_COLUMNS_IN_ISOLATION.key -> "false") {
+ // The normal iterations resolve at most one `withColumns` of the chain per iteration.
+ val message = intercept[RuntimeException](getAnalyzer.execute(chain)).getMessage
+ assert(message.startsWith("Max iterations (100) reached for batch Resolution"))
+ }
+ checkWithColumnsResolution(chain)
+ }
+
+ test("SPARK-59855: lazily planned withColumns that can't be resolved in isolation") {
Review Comment:
The added tests cover the 120/200 chains, the six "cannot resolve in isolation" cases, and error parity. How about adding cases that exercise the resolve-in-child vs. fall-back-to-normal split explicitly?
- a `withColumns` referencing a **metadata column** (e.g. `_metadata.file_path`);
- a Connect **plan-id reference** (`df[...]`) inside a chain (the description says this is resolved by `resolveReferencesInProject`; a test would pin it);
- a **SQL variable** reference;
- a **streaming** chain (the `isStreaming = p.child.isStreaming` path on the leaf);
- optionally, a `TempResolvedColumn`-like case, to confirm the isolated run never marks a not-fully-resolved expression as resolved. (In practice `TempResolvedColumn` is only produced under Aggregate/Filter/Sort, not under a `Project` over a `LocalRelation`, so this is likely already impossible but a test would document that.)
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -1997,6 +2010,60 @@ class Analyzer(
q.mapExpressions(resolveExpressionByPlanChildren(_, q, includeLastResort = true))
}
+ private def resolveReferencesInProject(p: Project): Project = {
+ val resolvedBasic = p.projectList.map(resolveExpressionByPlanChildren(_, p))
+ // Lateral column alias has higher priority than outer reference.
+ val resolvedWithLCA = resolveLateralColumnAlias(resolvedBasic)
+ val resolvedFinal = resolvedWithLCA.map(resolveColsLastResort)
+ p.copy(projectList =
+ resolvedFinal.map(e => aliasIfOuterReference(e.asInstanceOf[NamedExpression])))
+ }
+
+ /**
+ * Resolves a `withColumns` whose star has just been expanded.
+ *
+ * A lazily planned chain of `withColumns`, like the one Spark Connect creates for a chain of
+ * `withColumn` calls, is a stack of Projects. A Project waits for its child to be resolved,
+ * and then takes several rules to be resolved itself, so the normal iterations resolve at most
+ * one Project of the chain per iteration, and a long chain hits the max iterations limit.
+ * Instead, we resolve the Project right away: resolve its column references against the child
+ * as usual, then run the "Resolution" batch on the expressions that are still unresolved,
+ * over a leaf node with the same output as the child. This costs time proportional to the
+ * width of the Project instead of the size of the child, and lets the parent Project be
+ * resolved in the same pass. Expressions that can't be resolved this way, e.g. a window
+ * function that needs the plan to be rewritten, are left to the normal iterations.
+ */
+ private def resolveWithColumns(p: Project): Project = {
+ if (containsStar(p.projectList) || containsDeserializer(p.projectList) ||
+ p.projectList.exists(SubqueryExpression.hasSubquery)) {
+ p
+ } else {
+ val withReferences = resolveReferencesInProject(p)
+ val unresolved = withReferences.projectList.filterNot(_.resolved)
+ // Lateral column alias references need the whole project list to be resolved.
+ if (unresolved.isEmpty ||
+ unresolved.exists(_.containsPattern(LATERAL_COLUMN_ALIAS_REFERENCE))) {
+ withReferences
+ } else {
+ val leaf = LocalRelation(p.child.output, isStreaming = p.child.isStreaming)
+ try {
+ resolutionBatchExecutor.execute(Project(unresolved, leaf)) match {
+ case r @ Project(resolvedList, l) if (l eq leaf) && r.resolved &&
+ r.missingInput.isEmpty && resolvedList.length == unresolved.length =>
+ val resolvedIter = resolvedList.iterator
+ withReferences.copy(projectList = withReferences.projectList.map { e =>
+ if (e.resolved) e else resolvedIter.next()
+ })
+ case _ => withReferences
+ }
+ } catch {
+ // Leave the error to the normal iterations, so that it's reported as usual.
+ case _: AnalysisException => withReferences
Review Comment:
`resolveWithColumns` wraps the isolated run in `catch { case _: AnalysisException => withReferences }`. But when a batch exceeds `maxIterations`, `RuleExecutor.execute` throws a plain `RuntimeException`, not an `AnalysisException`.
https://github.com/apache/spark/blob/b03b53dcacb42c9c31c0cd2b622d495d60c3f59f/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleExecutor.scala#L299
The `Resolution` batch uses `fixedPoint` with `errorOnExceed = true`, and under `Utils.isTesting` the throw happens unconditionally. So if `resolutionBatchExecutor.execute(Project(unresolved, leaf))` does not converge within `maxIterations`, the `RuntimeException` escapes the `catch` and propagates to the user, instead of falling back to the normal iterations as the design intends. The same applies to non-`AnalysisException` errors that coercion rules can raise (e.g. `SparkException`).
The isolated run is a best-effort optimization whose failures should always fall back to the normal path. Suggest widening the catch:
```scala
catch { case NonFatal(_) => withReferences }
```
(or at least adding `RuntimeException`). Worth a test that forces the isolated run to not converge and asserts it falls back rather than surfacing a raw `RuntimeException`. Note the fallback here still runs the normal iterations, which for a genuinely divergent case may itself hit `maxIterations` but that is the pre-existing behavior and is the correct thing to reproduce.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -1997,6 +2010,60 @@ class Analyzer(
q.mapExpressions(resolveExpressionByPlanChildren(_, q, includeLastResort = true))
}
+ private def resolveReferencesInProject(p: Project): Project = {
+ val resolvedBasic = p.projectList.map(resolveExpressionByPlanChildren(_, p))
+ // Lateral column alias has higher priority than outer reference.
+ val resolvedWithLCA = resolveLateralColumnAlias(resolvedBasic)
+ val resolvedFinal = resolvedWithLCA.map(resolveColsLastResort)
+ p.copy(projectList =
+ resolvedFinal.map(e => aliasIfOuterReference(e.asInstanceOf[NamedExpression])))
+ }
+
+ /**
+ * Resolves a `withColumns` whose star has just been expanded.
+ *
+ * A lazily planned chain of `withColumns`, like the one Spark Connect creates for a chain of
+ * `withColumn` calls, is a stack of Projects. A Project waits for its child to be resolved,
+ * and then takes several rules to be resolved itself, so the normal iterations resolve at most
+ * one Project of the chain per iteration, and a long chain hits the max iterations limit.
+ * Instead, we resolve the Project right away: resolve its column references against the child
+ * as usual, then run the "Resolution" batch on the expressions that are still unresolved,
+ * over a leaf node with the same output as the child. This costs time proportional to the
+ * width of the Project instead of the size of the child, and lets the parent Project be
+ * resolved in the same pass. Expressions that can't be resolved this way, e.g. a window
+ * function that needs the plan to be rewritten, are left to the normal iterations.
+ */
+ private def resolveWithColumns(p: Project): Project = {
+ if (containsStar(p.projectList) || containsDeserializer(p.projectList) ||
+ p.projectList.exists(SubqueryExpression.hasSubquery)) {
+ p
+ } else {
+ val withReferences = resolveReferencesInProject(p)
+ val unresolved = withReferences.projectList.filterNot(_.resolved)
+ // Lateral column alias references need the whole project list to be resolved.
+ if (unresolved.isEmpty ||
+ unresolved.exists(_.containsPattern(LATERAL_COLUMN_ALIAS_REFERENCE))) {
+ withReferences
+ } else {
+ val leaf = LocalRelation(p.child.output, isStreaming = p.child.isStreaming)
+ try {
+ resolutionBatchExecutor.execute(Project(unresolved, leaf)) match {
+ case r @ Project(resolvedList, l) if (l eq leaf) && r.resolved &&
+ r.missingInput.isEmpty && resolvedList.length == unresolved.length =>
+ val resolvedIter = resolvedList.iterator
+ withReferences.copy(projectList = withReferences.projectList.map { e =>
Review Comment:
The re-splice is correct: `unresolved = withReferences.projectList.filterNot(_.resolved)` and the `map { if (e.resolved) e else resolvedIter.next() }` walk the same Seq in the same direction, so the i-th unresolved element maps to the i-th resolved element, and `resolvedList.length == unresolved.length` guards the count.
However, this relies on the `Resolution` batch preserving the order and count of the top-level project list of `Project(unresolved, leaf)` (the count is guarded, the order is not). A one-line comment stating that dependency would help, since a future rule that reorders the top-level list would silently break the mapping.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59855][SQL][CONNECT] Analyze long withColumns chains in one pass and remove the eager analysis in Spark Connect [spark]
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan closed pull request #59131: [SPARK-59855][SQL][CONNECT] Analyze long withColumns chains in one pass and remove the eager analysis in Spark Connect
URL: https://github.com/apache/spark/pull/59131
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-59855][SQL][CONNECT] Analyze long withColumns chains in one pass and remove the eager analysis in Spark Connect [spark]
Posted by "Vivek1106-04 (via GitHub)" <gi...@apache.org>.
Vivek1106-04 commented on PR #59131:
URL: https://github.com/apache/spark/pull/59131#issuecomment-5887440882
Thanks @cloud-fan, I tested this end to end against the SDP case from #59100.
**1. It fixes SPARK-59663.** The regression test from #59100 (`PythonPipelineSuite`, a `withColumn` / `withColumnRenamed` flow on `src` that already exists in the catalog from a previous run) passes on this PR. It might be worth carrying that test over here, since this PR has no pipelines test.
**2. Possible regression: `withColumn` interleaved with another operator in a loop.** The isolated resolution only applies to `Project(Seq(UnresolvedStarWithColumns))`. Any other operator in the chain whose expressions need a later rule (e.g. `ResolveFunctions`) still costs one fixed-point iteration. On master, the eager analysis of each `withColumns` also analyzed the whole child, so the operators in between were resolved along the way. With this PR, a loop like
```python
for i in range(1, n + 1):
df = df.withColumn(f"c{i}", F.col(prev) + 1).filter(F.col(f"c{i}").isNotNull())
```
now fails once `n` gets close to `spark.sql.analyzer.maxIterations`:
| n (withColumn + filter pairs) | master (eager) | this PR |
|---|---|---|
| 150 | 6.2 s | `Max iterations (100) reached for batch Resolution` |
| 1000 | 670 s | fails |
So master was slow (quadratic) but succeeded, while this PR fails with an error. The error is `HYBRID_ANALYZER_EXCEPTION.FIXED_POINT_FAILED_SINGLE_PASS_SUCCEEDED` in tests.
<details>
<summary>Repro (SparkConnectPlannerSuite)</summary>
```scala
test("withColumn interleaved with filter in a loop") {
def attr(name: String): proto.Expression =
proto.Expression
.newBuilder()
.setUnresolvedAttribute(
proto.Expression.UnresolvedAttribute.newBuilder().setUnparsedIdentifier(name))
.build()
def fn(name: String, args: proto.Expression*): proto.Expression =
proto.Expression
.newBuilder()
.setUnresolvedFunction(
proto.Expression.UnresolvedFunction
.newBuilder()
.setFunctionName(name)
.addAllArguments(args.asJava))
.build()
val one = proto.Expression.newBuilder().setLiteral(toLiteralProto(1L)).build()
val range = proto.Relation
.newBuilder()
.setRange(proto.Range.newBuilder().setStart(0).setEnd(10).setStep(1))
.build()
val n = 150
val chain = (1 to n).foldLeft(range) { (child, i) =>
val prev = attr(if (i == 1) "id" else s"c${i - 1}")
val withCol = proto.Relation
.newBuilder()
.setWithColumns(
proto.WithColumns
.newBuilder()
.setInput(child)
.addAliases(proto.Expression.Alias
.newBuilder()
.addName(s"c$i")
.setExpr(fn("+", prev, one))))
.build()
proto.Relation
.newBuilder()
.setFilter(
proto.Filter
.newBuilder()
.setInput(withCol)
.setCondition(fn("isnotnull", attr(s"c$i"))))
.build()
}
val df = Dataset.ofRows(spark, transform(chain))
assert(df.select(s"c$n").collect().map(_.getLong(0)).toSeq == (0L until 10L).map(_ + n))
}
```
</details>
**3. Same SDP bug remains for other eagerly analyzed relations (pre-existing, not introduced here).** The planner still analyzes `withWatermark`, `na.fill`, `na.drop` (and other `na` / `stat` relations) eagerly, so they have the same bind-at-build-time problem. On this PR, these flows still lose their dependency on `src` when `src` already exists in the catalog:
```python
@dp.materialized_view
def with_watermark():
return spark.read.table("src").withWatermark("ts", "1 minute")
@dp.materialized_view
def na_fill():
return spark.read.table("src").na.fill(0)
@dp.materialized_view
def na_drop():
return spark.read.table("src").na.drop()
```
```
List(with_watermark, na_fill, na_drop) was not empty flows lost their dependency on src
```
`withWatermark` is probably the most important for SDP since it's common in streaming flows. Happy to open a follow-up for these once this PR lands.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org