[SPARK-28346][SQL] clone the query plan between analyzer, optimizer and planner #25111

cloud-fan · 2019-07-11T06:44:39Z

What changes were proposed in this pull request?

query plan was designed to be immutable, but sometimes we do allow it to carry mutable states, because of the complexity of the SQL system. One example is TreeNodeTag. It's a state of TreeNode and can be carried over during copy and transform. The adaptive execution framework relies on it to link the logical and physical plans.

This leads to a problem: when we get QueryExecution#analyzed, the plan can be changed unexpectedly because it's mutable. I hit a real issue in #25107 : I use TreeNodeTag to carry dataset id in logical plans. However, the analyzed plan ends up with many duplicated dataset id tags in different nodes. It turns out that, the optimizer transforms the logical plan and add the tag to more nodes.

For example, the logical plan is SubqueryAlias(Filter(...)), and I expect only the SubqueryAlais has the dataset id tag. However, the optimizer removes SubqueryAlias and carries over the dataset id tag to Filter. When I go back to the analyzed plan, both SubqueryAlias and Filter has the dataset id tag, which breaks my assumption.

Since now query plan is mutable, I think it's better to limit the life cycle of a query plan instance. We can clone the query plan between analyzer, optimizer and planner, so that the life cycle is limited in one stage.

How was this patch tested?

new test

cloud-fan · 2019-07-11T06:48:44Z

...ain/scala/org/apache/spark/sql/catalyst/plans/logical/statsEstimation/LogicalPlanStats.scala

it's fragile to use member variable to keep stats, as they will be lost after copy.

cloud-fan · 2019-07-11T06:49:02Z

sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryRelation.scala

it's fragile to use member variable to keep stats, as they will be lost after copy.

cloud-fan · 2019-07-11T06:49:29Z

sql/core/src/main/scala/org/apache/spark/sql/execution/command/SetCommand.scala

The clone defined in TreeNode doesn't work for case object.

cloud-fan · 2019-07-11T06:50:11Z

...re/src/main/scala/org/apache/spark/sql/execution/datasources/SaveIntoDataSourceCommand.scala

The mapChildren in TreeNode will change the map type. (from CaseInsensitiveMap to a normal map)

cloud-fan · 2019-07-11T06:51:15Z

cc @hvanhovell @maryannxue @viirya @gatorsmile @HyukjinKwon

SparkQA · 2019-07-11T07:05:01Z

Test build #107513 has finished for PR 25111 at commit 656ae55.

This patch fails due to an unknown error code, -9.
This patch merges cleanly.
This patch adds no public classes.

cloud-fan · 2019-07-11T07:29:32Z

retest this please

SparkQA · 2019-07-11T09:55:40Z

Test build #107523 has finished for PR 25111 at commit 656ae55.

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

SparkQA · 2019-07-11T14:47:58Z

Test build #107534 has finished for PR 25111 at commit 7ab8e49.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2019-07-12T06:04:06Z

Test build #107573 has finished for PR 25111 at commit 58ff049.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

SparkQA · 2019-07-12T12:31:02Z

Test build #107587 has finished for PR 25111 at commit 92095b7.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

viirya · 2019-07-14T14:36:07Z

sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala

    sparkSession.sessionState.analyzer.executeAndCheck(logical, tracker)
  }

  lazy val withCachedData: LogicalPlan = {


Maybe not necessary, but should we clone logical too before sending to analyzer?

yea I think we should

viirya · 2019-07-14T14:55:57Z

sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryRelation.scala

-  @volatile var statsOfPlanToCache: Statistics = null
+  def getStatsOfPlanToCache(): Statistics = {
+    getTagValue(STATS_OF_PLAN_TO_CACHE_TAG).get
+  }


Hmm, statsOfPlanToCache has volatile semantics. But making it as a TreeNodeTag, seems we don't preserve that?

good point. tree node tag should be thread safe as well.

SparkQA · 2019-07-15T05:34:48Z

Test build #107657 has finished for PR 25111 at commit f614cc6.

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

SparkQA · 2019-07-15T16:38:55Z

Test build #107686 has finished for PR 25111 at commit 5e0ab9a.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

gatorsmile · 2019-07-16T16:40:44Z

cc @maryannxue @hvanhovell

SparkQA · 2019-07-16T21:05:10Z

Test build #107753 has finished for PR 25111 at commit 6f9b59f.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

maryannxue · 2019-07-18T02:50:06Z

The code looks good, but one thing we should be aware of:
clone() actually creates different copies/instances of MultiInstanceRelation objects, while before this change identical relations in a single query always share the same instance. This is what exactly needed by AQE, but would this contradict with any other existing assumptions (e.g., in datasource)?

cloud-fan · 2019-07-18T03:57:39Z

@maryannxue AFAIK we don't rely on plan instance equality in Spark. AQE is the only one I'm aware of that needs to check plan instance equality.

maryannxue · 2019-07-18T16:54:44Z

Thank you, @cloud-fan! LGTM.

gatorsmile · 2019-07-19T05:16:29Z

sql/core/src/main/scala/org/apache/spark/sql/execution/QueryExecution.scala


  lazy val optimizedPlan: LogicalPlan = tracker.measurePhase(QueryPlanningTracker.OPTIMIZATION) {
-    sparkSession.sessionState.optimizer.executeAndTrack(withCachedData, tracker)
+    sparkSession.sessionState.optimizer.executeAndTrack(withCachedData.clone(), tracker)


Since now query plan is mutable, I think it's better to limit the life cycle of a query plan instance. We can clone the query plan between analyzer, optimizer and planner, so that the life cycle is limited in one stage.

If we decide to clone the plan after each stage, will any test fail if we do not clone it?

SparkQA · 2019-07-19T18:17:52Z

Test build #107914 has finished for PR 25111 at commit 66f1281.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

gatorsmile · 2019-07-19T20:35:55Z

sql/core/src/test/scala/org/apache/spark/sql/execution/QueryExecutionSuite.scala

    assert(error.getMessage.contains("error"))
  }
+
+  test("analyzed plan should not change after it's generated") {


The tests still can pass without calling clone() in QueryExecution

cloud-fan · 2019-07-21T07:53:01Z

sql/core/src/test/scala/org/apache/spark/sql/execution/QueryExecutionSuite.scala

+    spark.experimental.extraStrategies = Nil
+  }
+
+  test("SPARK-28346: clone the query plan between analyzer, optimizer and planner") {


This test fails in the latest master branch.

SparkQA · 2019-07-23T11:42:48Z

Test build #108044 has finished for PR 25111 at commit 4f75ba4.

This patch passes all tests.
This patch merges cleanly.
This patch adds no public classes.

gatorsmile · 2019-07-23T15:58:07Z

LGTM

Thanks! Merged to master

### What changes were proposed in this pull request? Since JIRA SPARK-28346,PR [25111](#25111), QueryExecution will copy all node stage-by-stage. This make all node instance twice almost. So we should make all class fields lazy to avoid create more unexpected object. ### Why are the changes needed? Avoid create more unexpected object. ### Does this PR introduce any user-facing change? No. ### How was this patch tested? Exists UT. Closes #26565 from ulysses-you/make-val-lazy. Authored-by: ulysses <youxiduo@weidian.com> Signed-off-by: Dongjoon Hyun <dhyun@apple.com>

…nd planner query plan was designed to be immutable, but sometimes we do allow it to carry mutable states, because of the complexity of the SQL system. One example is `TreeNodeTag`. It's a state of `TreeNode` and can be carried over during copy and transform. The adaptive execution framework relies on it to link the logical and physical plans. This leads to a problem: when we get `QueryExecution#analyzed`, the plan can be changed unexpectedly because it's mutable. I hit a real issue in apache#25107 : I use `TreeNodeTag` to carry dataset id in logical plans. However, the analyzed plan ends up with many duplicated dataset id tags in different nodes. It turns out that, the optimizer transforms the logical plan and add the tag to more nodes. For example, the logical plan is `SubqueryAlias(Filter(...))`, and I expect only the `SubqueryAlais` has the dataset id tag. However, the optimizer removes `SubqueryAlias` and carries over the dataset id tag to `Filter`. When I go back to the analyzed plan, both `SubqueryAlias` and `Filter` has the dataset id tag, which breaks my assumption. Since now query plan is mutable, I think it's better to limit the life cycle of a query plan instance. We can clone the query plan between analyzer, optimizer and planner, so that the life cycle is limited in one stage. new test Closes apache#25111 from cloud-fan/clone. Authored-by: Wenchen Fan <wenchen@databricks.com> Signed-off-by: gatorsmile <gatorsmile@gmail.com>

MasterDDT · 2024-02-05T06:40:47Z

@cloud-fan I'm seeing some memory issues because of all these clone calls. I have a big query tree, maybe of ~20 height, so all the clone calls are recursive and keep everything in the stack alive: https://gist.github.com/MasterDDT/af98ad20ab0ed301476b9e8c58d8f5bb. 4g driver memory isnt enough on Spark 3.3, but I can run exact same workload on Spark 2.4 without any problems.

In my forked code, could I disable all the clone calls if AQE is off? Will that cause any correctness issues?

dongjoon-hyun · 2024-02-06T17:47:42Z

To @MasterDDT , I'd like to recommend to file an official JIRA issue. Otherwise, it's difficult to get any further discussion or help because this is too old thread.

cloud-fan changed the title ~~[SPARK-xxx][SQL] clone the query plan between analyzer, optimizer and planner~~ [SPARK-28346][SQL] clone the query plan between analyzer, optimizer and planner Jul 11, 2019

cloud-fan commented Jul 11, 2019

View reviewed changes

cloud-fan force-pushed the clone branch from 656ae55 to 7ab8e49 Compare July 11, 2019 11:21

dongjoon-hyun added the SQL label Jul 11, 2019

cloud-fan force-pushed the clone branch from 7ab8e49 to 58ff049 Compare July 12, 2019 02:28

clone the query plan between analyzer, optimizer and planner

92095b7

cloud-fan force-pushed the clone branch from 58ff049 to 92095b7 Compare July 12, 2019 09:06

viirya reviewed Jul 14, 2019

View reviewed changes

address comments

5e0ab9a

cloud-fan force-pushed the clone branch from f614cc6 to 5e0ab9a Compare July 15, 2019 13:22

simplify

6f9b59f

gatorsmile reviewed Jul 19, 2019

View reviewed changes

address comments

66f1281

viirya approved these changes Jul 19, 2019

View reviewed changes

gatorsmile reviewed Jul 19, 2019

View reviewed changes

cloud-fan commented Jul 21, 2019

View reviewed changes

improve test

4f75ba4

gatorsmile closed this in e04f696 Jul 23, 2019

ulysses-you mentioned this pull request Nov 18, 2019

[SPARK-29937][SQL] Make FileSourceScanExec class fields lazy #26565

Closed

cloud-fan mentioned this pull request Jan 31, 2020

backport [SPARK-27747][SPARK-27816][SPARK-28344] #27417

Closed

[SPARK-28346][SQL] clone the query plan between analyzer, optimizer and planner #25111

[SPARK-28346][SQL] clone the query plan between analyzer, optimizer and planner #25111

Uh oh!

Conversation

cloud-fan commented Jul 11, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

How was this patch tested?

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cloud-fan Jul 11, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cloud-fan commented Jul 11, 2019

Uh oh!

SparkQA commented Jul 11, 2019

Uh oh!

cloud-fan commented Jul 11, 2019

Uh oh!

SparkQA commented Jul 11, 2019

Uh oh!

SparkQA commented Jul 11, 2019

Uh oh!

SparkQA commented Jul 12, 2019

Uh oh!

SparkQA commented Jul 12, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

viirya Jul 14, 2019 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Jul 15, 2019

Uh oh!

SparkQA commented Jul 15, 2019

Uh oh!

gatorsmile commented Jul 16, 2019

Uh oh!

SparkQA commented Jul 16, 2019

Uh oh!

maryannxue commented Jul 18, 2019

Uh oh!

cloud-fan commented Jul 18, 2019

Uh oh!

maryannxue commented Jul 18, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Jul 19, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

SparkQA commented Jul 23, 2019

Uh oh!

gatorsmile commented Jul 23, 2019

Uh oh!

MasterDDT commented Feb 5, 2024 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

dongjoon-hyun commented Feb 6, 2024

Uh oh!

Reviewers

Assignees

Labels

Projects

cloud-fan commented Jul 11, 2019 •

edited

Loading

cloud-fan Jul 11, 2019 •

edited

Loading

viirya Jul 14, 2019 •

edited

Loading

MasterDDT commented Feb 5, 2024 •

edited

Loading