[SPARK-28548][SQL] explain() shows wrong result for persisted DataFrames after some operations

[sarutak]

What changes were proposed in this pull request?

After some operations against Datasets and then persist them, Dataset.explain shows wrong result.
One of those operations is explain() itself.
An example here.

val df = spark.range(10)
df.explain
df.persist
df.explain

Expected result is like as follows.

== Physical Plan ==
*(1) ColumnarToRow
+- InMemoryTableScan [id#7L]
      +- InMemoryRelation [id#7L], StorageLevel(disk, memory, deserialized, 1 replicas)
            +- *(1) Range (0, 10, step=1, splits=12)

But I got this.

== Physical Plan ==
*(1) Range (0, 10, step=1, splits=12)

This issue is caused by withCachedData in QueryExecution is materialized early when explain() or such methods are called so this patch prevents it.

How was this patch tested?

Additional test cases in ExplainSuite.scala

[SparkQA]

Test build #108278 has finished for PR 25280 at commit 3b16d9a.

• This patch fails Scala style tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #108279 has finished for PR 25280 at commit e76e2b3.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[maropu]

wait, it seems the query works well in v2.4.3;

scala> df.explain
== Physical Plan ==
*(1) Range (0, 10, step=1, splits=4)

scala> df.persist
res1: df.type = [id: bigint]

scala> df.explain
== Physical Plan ==
*(1) InMemoryTableScan [id#0L]
   +- InMemoryRelation [id#0L], StorageLevel(disk, memory, deserialized, 1 replicas)
         +- *(1) Range (0, 10, step=1, splits=4)
scala> 

Which commit affects the behaviour?

[cloud-fan]

It's by design that a dataframe can't change its physical plan once the physical plan is materialized. That said, df.persist has no effect if df.explain is called before.

I think the current behavior is corrected and v2.4.3 is wrong. I think this problem is fixed by #24654

If you do wanna run the plan with cached data, maybe we can do val df2 = df.toDF() and execute df2 instead.

[cloud-fan]

Actually, this does expose a problem. Before #24654, df.explain won't materialize the physical plan, but now we do. cc @viirya do you have any ideas? I think this one is hard to fix.

[sarutak]

@cloud-fan I have already noticed #24654. The problem mentioned in that PR is that pre-analyzed logical plan was always reused and re-analyzed in explain command even though analyzed-logical plan is already materialized.

This solution consider that problem. If we have already analyzed-logical plan materialized, we use it otherwise create one.

[viirya]

We actually take the query execution of dataset to execute. If we have executed a dataset, so its physical plan is materialized, then persist it. In 2.4.3, although df.explain shows a cached plan, I think execution still uses physical plan without cache? This fix also has the issue?

That said, in 2.4.3, df.explain shows query plan with current status like cache, temp view, I think it doesn't really match with dataset execution.

Like:

val df = spark.range(10)
df.explain // show query plan without cache
df.collect() // execution without cache 
df.persist
df.explain // show query plan with cache
df.collect() // still execution without cache

[SparkQA]

Test build #108331 has finished for PR 25280 at commit 98bcee4.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
• case class Plans(

[maropu]

We cannot just return a new Dataset in persist, btw, like this?

val df = spark.range(10)
df.explain // show query plan without cache
df.collect // execution without cache

val cachedDf = df.persist

df.explain // show query plan without cache
df.collect // execution without cache

cachedDf.explain // show query plan with cache
cachedDf.collect // execution with cache

[cloud-fan]

@maropu this is a behavior change and can break many queries silently.

[sarutak]

@viirya With my change, we can get following result.

val df = spark.range(10)
df.explain  // show query plan without cache
df.collect  // execution without cache
df.persist
df.explain // show query plan without cache
df.collect // execution without cache
df.queryExecution.executedPlan.find(_.isInstanceOf[InMemoryTableScanExec]) // None

After collect, persist is still ignored but this result is different from one of 2.4.3 and same for master branch.
As you mentioned df.collect and some operations materialize the executedPlan and cause this problem.

Some operations including df.show don't cause this problem because they create the new root plan at the time of execution implicitly so I wonder creating dummy root plan at the time of executing collect or similar operations resolve this type of problem?

[felixcheung]

... where are we on this? this seems to be severe correctness impacting issue?

[cloud-fan]

This is not a correctness issue. This is just an undocumented property "a dataframe can't change its physical plan once the physical plan is materialized" causing confusions.

After so many days, I have a fresh idea now: when df.cache is called, reset the cachedPlan, optimizedPlan and physical plans. We can implement the lazy evaluation manually to support it instead of relying on Scals lazy val.

[github-actions]

We're closing this PR because it hasn't been updated in a while.
This isn't a judgement on the merit of the PR in any way. It's just
a way of keeping the PR queue manageable.

If you'd like to revive this PR, please reopen it!

[maropu]

btw, we don't need to describe something about this behaviour in the migration guide if the current one is correct? @gatorsmile @cloud-fan

[HyukjinKwon]

I wouldn't mind documenting it but sounds more like a bug fix given my reading, and it seems only affecting a debug API (explain). So I suspect it's fine not to document for now.

[maropu]

ok, thanks for the check.