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 2023/08/03 06:17:57 UTC
[GitHub] [spark] cloud-fan opened a new pull request, #42315: [SPARK-44653][SQL] Non-trivial DataFrame unions should not break caching
cloud-fan opened a new pull request, #42315:
URL: https://github.com/apache/spark/pull/42315
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://spark.apache.org/contributing.html
2. Ensure you have added or run the appropriate tests for your PR: https://spark.apache.org/developer-tools.html
3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][SPARK-XXXX] Your PR title ...'.
4. Be sure to keep the PR description updated to reflect all changes.
5. Please write your PR title to summarize what this PR proposes.
6. If possible, provide a concise example to reproduce the issue for a faster review.
7. If you want to add a new configuration, please read the guideline first for naming configurations in
'core/src/main/scala/org/apache/spark/internal/config/ConfigEntry.scala'.
8. If you want to add or modify an error type or message, please read the guideline first in
'core/src/main/resources/error/README.md'.
-->
### What changes were proposed in this pull request?
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
We have a long-standing tricky optimization in `Dataset.union`, which invokes the optimizer rule `CombineUnions` to pre-optimize the analyzed plan. This is to avoid too large analyzed plan for a specific dataframe query pattern `df1.union(df2).union(df3).union...`.
This tricky optimization is designed to break dataframe caching, but we thought it was fine as people usually won't cache the intermediate dataframe in a union chain. However, `CombineUnions` gets improved from time to time and now it can optimize a wide range of Union patterns. Now it's possible that people union two dataframe, do something with `select`, and cache it. Then the dataframe is unioned again with other dataframes and people expect the df cache to work. However the cache won't work due to the tricky optimization in `Dataset.union`.
This PR updates `Dataset.union` to only combine adjacent Unions to match the original purpose.
### Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
Fix perf regression due to breaking df caching
### Does this PR introduce _any_ user-facing change?
<!--
Note that it means *any* user-facing change including all aspects such as the documentation fix.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description and/or an example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Spark versions or within the unreleased branches such as master.
If no, write 'No'.
-->
no
### How was this patch tested?
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
If benchmark tests were added, please run the benchmarks in GitHub Actions for the consistent environment, and the instructions could accord to: https://spark.apache.org/developer-tools.html#github-workflow-benchmarks.
-->
new test
--
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
[GitHub] [spark] wangyum commented on a diff in pull request #42315: [SPARK-44653][SQL] Non-trivial DataFrame unions should not break caching
Posted by "wangyum (via GitHub)" <gi...@apache.org>.
wangyum commented on code in PR #42315:
URL: https://github.com/apache/spark/pull/42315#discussion_r1283918802
##########
sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala:
##########
@@ -2272,9 +2316,7 @@ class Dataset[T] private[sql](
* @since 2.0.0
*/
def union(other: Dataset[T]): Dataset[T] = withSetOperator {
- // This breaks caching, but it's usually ok because it addresses a very specific use case:
- // using union to union many files or partitions.
- CombineUnions(Union(logicalPlan, other.logicalPlan))
Review Comment:
Could we add a flag to `CombineUnions`?
```patch
Index: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
IDEA additional info:
Subsystem: com.intellij.openapi.diff.impl.patch.CharsetEP
<+>UTF-8
===================================================================
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala (revision d9d5325eed228c61d0825160477d0defaabee710)
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala (date 1691116492482)
@@ -91,7 +91,7 @@
CollapseWindow,
EliminateOffsets,
EliminateLimits,
- CombineUnions,
+ CombineUnions(),
// Constant folding and strength reduction
OptimizeRepartition,
TransposeWindow,
@@ -159,7 +159,7 @@
InlineCTE()) ::
Batch("Union", Once,
RemoveNoopOperators,
- CombineUnions,
+ CombineUnions(),
RemoveNoopUnion) ::
// Run this once earlier. This might simplify the plan and reduce cost of optimizer.
// For example, a query such as Filter(LocalRelation) would go through all the heavy
@@ -1451,7 +1451,7 @@
/**
* Combines all adjacent [[Union]] operators into a single [[Union]].
*/
-object CombineUnions extends Rule[LogicalPlan] {
+case class CombineUnions(combineWithProject: Boolean = true) extends Rule[LogicalPlan] {
import CollapseProject.{buildCleanedProjectList, canCollapseExpressions}
import PushProjectionThroughUnion.pushProjectionThroughUnion
@@ -1480,7 +1480,8 @@
while (stack.nonEmpty) {
stack.pop() match {
case p1 @ Project(_, p2: Project)
- if canCollapseExpressions(p1.projectList, p2.projectList, alwaysInline = false) &&
+ if combineWithProject &&
+ canCollapseExpressions(p1.projectList, p2.projectList, alwaysInline = false) &&
!p1.projectList.exists(SubqueryExpression.hasCorrelatedSubquery) &&
!p2.projectList.exists(SubqueryExpression.hasCorrelatedSubquery) =>
val newProjectList = buildCleanedProjectList(p1.projectList, p2.projectList)
@@ -1499,15 +1500,17 @@
// Push down projection through Union and then push pushed plan to Stack if
// there is a Project.
case Project(projectList, Distinct(u @ Union(children, byName, allowMissingCol)))
- if projectList.forall(_.deterministic) && children.nonEmpty &&
+ if combineWithProject && projectList.forall(_.deterministic) && children.nonEmpty &&
flattenDistinct && byName == topByName && allowMissingCol == topAllowMissingCol =>
stack.pushAll(pushProjectionThroughUnion(projectList, u).reverse)
case Project(projectList, Deduplicate(keys: Seq[Attribute], u: Union))
- if projectList.forall(_.deterministic) && flattenDistinct && u.byName == topByName &&
+ if combineWithProject &&
+ projectList.forall(_.deterministic) && flattenDistinct && u.byName == topByName &&
u.allowMissingCol == topAllowMissingCol && AttributeSet(keys) == u.outputSet =>
stack.pushAll(pushProjectionThroughUnion(projectList, u).reverse)
case Project(projectList, u @ Union(children, byName, allowMissingCol))
- if projectList.forall(_.deterministic) && children.nonEmpty &&
+ if combineWithProject &&
+ projectList.forall(_.deterministic) && children.nonEmpty &&
byName == topByName && allowMissingCol == topAllowMissingCol =>
stack.pushAll(pushProjectionThroughUnion(projectList, u).reverse)
case child =>
```
--
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
[GitHub] [spark] wangyum commented on a diff in pull request #42315: [SPARK-44653][SQL] Non-trivial DataFrame unions should not break caching
Posted by "wangyum (via GitHub)" <gi...@apache.org>.
wangyum commented on code in PR #42315:
URL: https://github.com/apache/spark/pull/42315#discussion_r1282789589
##########
sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala:
##########
@@ -2241,6 +2240,18 @@ class Dataset[T] private[sql](
Offset(Literal(n), logicalPlan)
}
+ // This breaks caching, but it's usually ok because it addresses a very specific use case:
+ // using union to union many files or partitions.
+ private def combineUnions(u: Union): LogicalPlan = {
+ val newChildren = u.children.flatMap {
+ case child: Union if u.byName == child.byName && u.allowMissingCol == child.allowMissingCol =>
+ child.children
+ case other =>
+ Seq(other)
+ }
+ u.withNewChildren(newChildren)
Review Comment:
`withNewChildren` needs to ensure the size of children is consistent. Can we use `Union(newChildren)`?
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #42315: [SPARK-44653][SQL] Non-trivial DataFrame unions should not break caching
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #42315:
URL: https://github.com/apache/spark/pull/42315#discussion_r1283926955
##########
sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala:
##########
@@ -2272,9 +2316,7 @@ class Dataset[T] private[sql](
* @since 2.0.0
*/
def union(other: Dataset[T]): Dataset[T] = withSetOperator {
- // This breaks caching, but it's usually ok because it addresses a very specific use case:
- // using union to union many files or partitions.
- CombineUnions(Union(logicalPlan, other.logicalPlan))
Review Comment:
I want to be future-proof. We may add more improvements to `CombineUnions` in the future and I don't want to evaluate the dataframe impact every time.
--
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
[GitHub] [spark] cloud-fan closed pull request #42315: [SPARK-44653][SQL] Non-trivial DataFrame unions should not break caching
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan closed pull request #42315: [SPARK-44653][SQL] Non-trivial DataFrame unions should not break caching
URL: https://github.com/apache/spark/pull/42315
--
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
[GitHub] [spark] cloud-fan commented on a diff in pull request #42315: [SPARK-44653][SQL] Non-trivial DataFrame unions should not break caching
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on code in PR #42315:
URL: https://github.com/apache/spark/pull/42315#discussion_r1283875186
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -157,7 +157,7 @@ abstract class Optimizer(catalogManager: CatalogManager)
// since the other rules might make two separate Unions operators adjacent.
Batch("Inline CTE", Once,
InlineCTE()) ::
- Batch("Union", Once,
+ Batch("Union", fixedPoint,
Review Comment:
Without the pre-optimization in the dataframe side, some pyspark tests failed complaining that this batch is not idempotent. In fact, this optimization batch does not need to be idempotent, we use `Once` before because we thought it was sufficient, but it's actually not sufficient.
--
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
[GitHub] [spark] wangyum commented on a diff in pull request #42315: [SPARK-44653][SQL] Non-trivial DataFrame unions should not break caching
Posted by "wangyum (via GitHub)" <gi...@apache.org>.
wangyum commented on code in PR #42315:
URL: https://github.com/apache/spark/pull/42315#discussion_r1283936632
##########
sql/core/src/main/scala/org/apache/spark/sql/Dataset.scala:
##########
@@ -2272,9 +2316,7 @@ class Dataset[T] private[sql](
* @since 2.0.0
*/
def union(other: Dataset[T]): Dataset[T] = withSetOperator {
- // This breaks caching, but it's usually ok because it addresses a very specific use case:
- // using union to union many files or partitions.
- CombineUnions(Union(logicalPlan, other.logicalPlan))
Review Comment:
OK.
--
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
[GitHub] [spark] viirya commented on a diff in pull request #42315: [SPARK-44653][SQL] Non-trivial DataFrame unions should not break caching
Posted by "viirya (via GitHub)" <gi...@apache.org>.
viirya commented on code in PR #42315:
URL: https://github.com/apache/spark/pull/42315#discussion_r1283470629
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -157,7 +157,7 @@ abstract class Optimizer(catalogManager: CatalogManager)
// since the other rules might make two separate Unions operators adjacent.
Batch("Inline CTE", Once,
InlineCTE()) ::
- Batch("Union", Once,
+ Batch("Union", fixedPoint,
Review Comment:
Why change to fixed point?
--
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
[GitHub] [spark] cloud-fan commented on pull request #42315: [SPARK-44653][SQL] Non-trivial DataFrame unions should not break caching
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #42315:
URL: https://github.com/apache/spark/pull/42315#issuecomment-1663353820
cc @wangyum @viirya @tanelk @allisonwang-db
--
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
[GitHub] [spark] cloud-fan commented on pull request #42315: [SPARK-44653][SQL] Non-trivial DataFrame unions should not break caching
Posted by "cloud-fan (via GitHub)" <gi...@apache.org>.
cloud-fan commented on PR #42315:
URL: https://github.com/apache/spark/pull/42315#issuecomment-1664918205
thanks for the review, merging to master/3.5/3.4!
--
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