[SPARK-41183][SQL] Add an extension API to do plan normalization for caching #38692
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Contributor
Author
viirya
reviewed
Nov 18, 2022
| * <ul> | ||
| * <li>Analyzer Rules.</li> | ||
| * <li>Check Analysis Rules.</li> | ||
| * <li>Plan Normalization Rules.</li> |
Member
There was a problem hiding this comment.
Cache Plan Normalization Rules? Since it is used for cache only.
viirya
reviewed
Nov 18, 2022
| case other => other | ||
| } | ||
|
|
||
| lazy val normalized: LogicalPlan = { |
Member
There was a problem hiding this comment.
Perhaps adding some comments for this.
viirya
reviewed
Nov 18, 2022
Comment on lines
+204
to
+207
| df.select("i").filter($"i" > 1).cache() | ||
| assert(df.filter($"i" > 1).select("i").queryExecution.executedPlan.find { | ||
| case _: org.apache.spark.sql.execution.columnar.InMemoryTableScanExec => true | ||
| case _ => false |
Member
There was a problem hiding this comment.
So without the added rule, caching is unable to apply here, right?
Contributor
There was a problem hiding this comment.
Should we add a negative test that verifies this? Might be overkill...
viirya
reviewed
Nov 18, 2022
| planChangeLogger.logRule(rule.ruleName, p, result) | ||
| result | ||
| }) | ||
| if (normalizationRules.nonEmpty) { |
Member
There was a problem hiding this comment.
I think entering this else block means normalizationRules is non empty, no?
| private[this] val planNormalizationRules = mutable.Buffer.empty[RuleBuilder] | ||
|
|
||
| def buildPlanNormalizationRules(session: SparkSession): Seq[Rule[LogicalPlan]] = { | ||
| planNormalizationRules.map(_.apply(session)).toSeq |
Contributor
There was a problem hiding this comment.
nit: isn't .apply(...) redundant with just (...) ?
Contributor
Author
There was a problem hiding this comment.
I'm following the existing code style in this file. I assume the reason is people who are not familiar with Scala may be confused when reading the code .map(_(session))
| commandExecuted | ||
| } else { | ||
| val planChangeLogger = new PlanChangeLogger[LogicalPlan]() | ||
| val normalized = normalizationRules.foldLeft(commandExecuted)((p, rule) => { |
Contributor
There was a problem hiding this comment.
nit: (... => { ... }) is equivalent to just { ... => ... } in this context
viirya
approved these changes
Nov 18, 2022
Contributor
Author
|
thanks for the review, merging to master! |
Member
dongjoon-hyun
left a comment
dongjoon-hyun
left a comment
There was a problem hiding this comment.
+1, late LGTM. Thank you, @cloud-fan and all.
HyukjinKwon
pushed a commit
that referenced
this pull request
Nov 24, 2022
…tionRules to injectPlanNormalizationRule ### What changes were proposed in this pull request? Followup of #38692. To follow other APIs in `SparkSessionExtensions`, the name should be `inject...Rule` and `build...Rules`. ### Why are the changes needed? typo fix ### Does this PR introduce _any_ user-facing change? not a released API ### How was this patch tested? n/a Closes #38767 from cloud-fan/small. Authored-by: Wenchen Fan <[email protected]> Signed-off-by: Hyukjin Kwon <[email protected]>
SandishKumarHN
pushed a commit
to SandishKumarHN/spark
that referenced
this pull request
Dec 12, 2022
…caching ### What changes were proposed in this pull request? Today, Spark is very conservative and uses the analyzed plan instead of the optimized plan as the cache key. Many cache opportunities are missed. This PR updates `SparkSessionExtensions` to allow people to inject custom plan normalization rules. Users can pick some safe optimizer rules, or implement new rules based on their business needs, to do plan normalization and increase the cache hit rate. ### Why are the changes needed? allow advanced users to do caching better. ### Does this PR introduce _any_ user-facing change? no ### How was this patch tested? new test Closes apache#38692 from cloud-fan/cache. Authored-by: Wenchen Fan <[email protected]> Signed-off-by: Wenchen Fan <[email protected]>
beliefer
pushed a commit
to beliefer/spark
that referenced
this pull request
Dec 15, 2022
…caching ### What changes were proposed in this pull request? Today, Spark is very conservative and uses the analyzed plan instead of the optimized plan as the cache key. Many cache opportunities are missed. This PR updates `SparkSessionExtensions` to allow people to inject custom plan normalization rules. Users can pick some safe optimizer rules, or implement new rules based on their business needs, to do plan normalization and increase the cache hit rate. ### Why are the changes needed? allow advanced users to do caching better. ### Does this PR introduce _any_ user-facing change? no ### How was this patch tested? new test Closes apache#38692 from cloud-fan/cache. Authored-by: Wenchen Fan <[email protected]> Signed-off-by: Wenchen Fan <[email protected]>
beliefer
pushed a commit
to beliefer/spark
that referenced
this pull request
Dec 15, 2022
…tionRules to injectPlanNormalizationRule ### What changes were proposed in this pull request? Followup of apache#38692. To follow other APIs in `SparkSessionExtensions`, the name should be `inject...Rule` and `build...Rules`. ### Why are the changes needed? typo fix ### Does this PR introduce _any_ user-facing change? not a released API ### How was this patch tested? n/a Closes apache#38767 from cloud-fan/small. Authored-by: Wenchen Fan <[email protected]> Signed-off-by: Hyukjin Kwon <[email protected]>
beliefer
pushed a commit
to beliefer/spark
that referenced
this pull request
Dec 18, 2022
…caching ### What changes were proposed in this pull request? Today, Spark is very conservative and uses the analyzed plan instead of the optimized plan as the cache key. Many cache opportunities are missed. This PR updates `SparkSessionExtensions` to allow people to inject custom plan normalization rules. Users can pick some safe optimizer rules, or implement new rules based on their business needs, to do plan normalization and increase the cache hit rate. ### Why are the changes needed? allow advanced users to do caching better. ### Does this PR introduce _any_ user-facing change? no ### How was this patch tested? new test Closes apache#38692 from cloud-fan/cache. Authored-by: Wenchen Fan <[email protected]> Signed-off-by: Wenchen Fan <[email protected]>
beliefer
pushed a commit
to beliefer/spark
that referenced
this pull request
Dec 18, 2022
…tionRules to injectPlanNormalizationRule ### What changes were proposed in this pull request? Followup of apache#38692. To follow other APIs in `SparkSessionExtensions`, the name should be `inject...Rule` and `build...Rules`. ### Why are the changes needed? typo fix ### Does this PR introduce _any_ user-facing change? not a released API ### How was this patch tested? n/a Closes apache#38767 from cloud-fan/small. Authored-by: Wenchen Fan <[email protected]> Signed-off-by: Hyukjin Kwon <[email protected]>
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
4 participants
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this pull request?
Today, Spark is very conservative and uses the analyzed plan instead of the optimized plan as the cache key. Many cache opportunities are missed.
This PR updates
SparkSessionExtensionsto allow people to inject custom plan normalization rules. Users can pick some safe optimizer rules, or implement new rules based on their business needs, to do plan normalization and increase the cache hit rate.Why are the changes needed?
allow advanced users to do caching better.
Does this PR introduce any user-facing change?
no
How was this patch tested?
new test