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: [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]
