You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2021/05/19 14:10:31 UTC
[GitHub] [spark] Ngone51 opened a new pull request #32590: [SPARK-35445][SQL] Reduce the execution time of DeduplicateRelations
Ngone51 opened a new pull request #32590:
URL: https://github.com/apache/spark/pull/32590
<!--
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 message, please read the guideline first:
https://spark.apache.org/error-message-guidelines.html
-->
### 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.
-->
This PR reduces the execution time of `DeduplicateRelations` by:
1) use `Set` instead `Seq` to check duplicate relations
2) avoid plan rewrites when there's no changes in children plan
### 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.
-->
Rule `DeduplicateRelations` is slow.
### 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.
-->
Run `TPCDSQuerySuite` and checked the run time of `DeduplicateRelations`. The time has been reduced by 77.9% after this PR.
--
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.
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] SparkQA commented on pull request #32590: [SPARK-35445][SQL] Reduce the execution time of DeduplicateRelations
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32590:
URL: https://github.com/apache/spark/pull/32590#issuecomment-844389821
**[Test build #138712 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/138712/testReport)** for PR 32590 at commit [`f085918`](https://github.com/apache/spark/commit/f0859185cad086a8a7fe4732cc3ca1de3834bfee).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds the following public classes _(experimental)_:
* `case class ReferenceEqualPlanWrapper(plan: LogicalPlan) `
--
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.
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] AmplabJenkins commented on pull request #32590: [SPARK-35445][SQL] Reduce the execution time of DeduplicateRelations
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32590:
URL: https://github.com/apache/spark/pull/32590#issuecomment-845144933
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/138737/
--
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.
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] Ngone51 commented on a change in pull request #32590: [SPARK-35445][SQL] Reduce the execution time of DeduplicateRelations
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on a change in pull request #32590:
URL: https://github.com/apache/spark/pull/32590#discussion_r635895661
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/DeduplicateRelations.scala
##########
@@ -17,16 +17,30 @@
package org.apache.spark.sql.catalyst.analysis
-import scala.collection.mutable.ArrayBuffer
+import scala.collection.mutable
import org.apache.spark.sql.catalyst.expressions.{Alias, AttributeMap, AttributeSet, NamedExpression, SubqueryExpression}
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.trees.AlwaysProcess
+/**
+ * A help class that used to detect duplicate relations fast in `DeduplicateRelations`
+ */
+case class ReferenceEqualPlanWrapper(plan: LogicalPlan) {
+ private val _hashCode = System.identityHashCode(plan)
Review comment:
Yea..I actually tried `IdentityHasMap` too but it doesn't help a lot...
--
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.
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] SparkQA commented on pull request #32590: [SPARK-35445][SQL] Reduce the execution time of DeduplicateRelations
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32590:
URL: https://github.com/apache/spark/pull/32590#issuecomment-845110940
**[Test build #138737 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/138737/testReport)** for PR 32590 at commit [`ef50b77`](https://github.com/apache/spark/commit/ef50b77b580e3b9a2ed57b8bc22258569f596212).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
--
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.
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] Ngone51 commented on pull request #32590: [SPARK-35445][SQL] Reduce the execution time of DeduplicateRelations
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on pull request #32590:
URL: https://github.com/apache/spark/pull/32590#issuecomment-845665122
thanks all!
--
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.
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] AmplabJenkins removed a comment on pull request #32590: [SPARK-35445][SQL] Reduce the execution time of DeduplicateRelations
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32590:
URL: https://github.com/apache/spark/pull/32590#issuecomment-844978646
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/43260/
--
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.
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] AmplabJenkins removed a comment on pull request #32590: [SPARK-35445][SQL] Reduce the execution time of DeduplicateRelations
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32590:
URL: https://github.com/apache/spark/pull/32590#issuecomment-844284854
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/43233/
--
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.
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] AmplabJenkins removed a comment on pull request #32590: [SPARK-35445][SQL] Reduce the execution time of DeduplicateRelations
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32590:
URL: https://github.com/apache/spark/pull/32590#issuecomment-844419653
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/138712/
--
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.
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] SparkQA commented on pull request #32590: [SPARK-35445][SQL] Reduce the execution time of DeduplicateRelations
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32590:
URL: https://github.com/apache/spark/pull/32590#issuecomment-844928082
Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/43260/
--
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.
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] AmplabJenkins commented on pull request #32590: [SPARK-35445][SQL] Reduce the execution time of DeduplicateRelations
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32590:
URL: https://github.com/apache/spark/pull/32590#issuecomment-845867966
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/138782/
--
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.
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] Ngone51 commented on a change in pull request #32590: [SPARK-35445][SQL] Reduce the execution time of DeduplicateRelations
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on a change in pull request #32590:
URL: https://github.com/apache/spark/pull/32590#discussion_r635358225
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/DeduplicateRelations.scala
##########
@@ -17,16 +17,30 @@
package org.apache.spark.sql.catalyst.analysis
-import scala.collection.mutable.ArrayBuffer
+import scala.collection.mutable
import org.apache.spark.sql.catalyst.expressions.{Alias, AttributeMap, AttributeSet, NamedExpression, SubqueryExpression}
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.trees.AlwaysProcess
+/**
+ * A help class that used to detect duplicate relations fast in `DeduplicateRelations`
+ */
+case class ReferenceEqualPlanWrapper(plan: LogicalPlan) {
+ private val _hashCode = System.identityHashCode(plan)
Review comment:
We only want to know if the two plans are the same instance. By default, `System.identityHashCode` returns the memory address of an object, which is faster than plan.hash() (which involves more details about the plan itself.)
--
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.
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] AmplabJenkins commented on pull request #32590: [SPARK-35445][SQL] Reduce the execution time of DeduplicateRelations
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32590:
URL: https://github.com/apache/spark/pull/32590#issuecomment-844978646
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/43260/
--
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.
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] AmplabJenkins commented on pull request #32590: [SPARK-35445][SQL] Reduce the execution time of DeduplicateRelations
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32590:
URL: https://github.com/apache/spark/pull/32590#issuecomment-844419653
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/138712/
--
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.
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] SparkQA commented on pull request #32590: [SPARK-35445][SQL] Reduce the execution time of DeduplicateRelations
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32590:
URL: https://github.com/apache/spark/pull/32590#issuecomment-844224171
Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/43233/
--
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.
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] AmplabJenkins commented on pull request #32590: [SPARK-35445][SQL] Reduce the execution time of DeduplicateRelations
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32590:
URL: https://github.com/apache/spark/pull/32590#issuecomment-844284854
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/43233/
--
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.
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] gengliangwang commented on a change in pull request #32590: [SPARK-35445][SQL] Reduce the execution time of DeduplicateRelations
Posted by GitBox <gi...@apache.org>.
gengliangwang commented on a change in pull request #32590:
URL: https://github.com/apache/spark/pull/32590#discussion_r635340026
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/DeduplicateRelations.scala
##########
@@ -17,16 +17,30 @@
package org.apache.spark.sql.catalyst.analysis
-import scala.collection.mutable.ArrayBuffer
+import scala.collection.mutable
import org.apache.spark.sql.catalyst.expressions.{Alias, AttributeMap, AttributeSet, NamedExpression, SubqueryExpression}
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.trees.AlwaysProcess
+/**
+ * A help class that used to detect duplicate relations fast in `DeduplicateRelations`
Review comment:
nit: helper
--
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.
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] gengliangwang commented on a change in pull request #32590: [SPARK-35445][SQL] Reduce the execution time of DeduplicateRelations
Posted by GitBox <gi...@apache.org>.
gengliangwang commented on a change in pull request #32590:
URL: https://github.com/apache/spark/pull/32590#discussion_r635353908
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/DeduplicateRelations.scala
##########
@@ -67,61 +81,78 @@ object DeduplicateRelations extends Rule[LogicalPlan] {
* all relations of the new LogicalPlan )
*/
private def renewDuplicatedRelations(
- existingRelations: Seq[MultiInstanceRelation],
- plan: LogicalPlan): (LogicalPlan, Seq[MultiInstanceRelation]) = plan match {
- case p: LogicalPlan if p.isStreaming => (plan, Nil)
+ existingRelations: mutable.HashSet[ReferenceEqualPlanWrapper],
+ plan: LogicalPlan)
+ : (LogicalPlan, mutable.HashSet[ReferenceEqualPlanWrapper], Boolean) = plan match {
+ case p: LogicalPlan if p.isStreaming => (plan, mutable.HashSet.empty, false)
case m: MultiInstanceRelation =>
- if (isDuplicated(existingRelations, m)) {
+ val planWrapper = ReferenceEqualPlanWrapper(m)
+ if (isDuplicated(existingRelations, planWrapper)) {
val newNode = m.newInstance()
newNode.copyTagsFrom(m)
- (newNode, Nil)
+ (newNode, mutable.HashSet.empty, true)
} else {
- (m, Seq(m))
+ val mWrapper = new mutable.HashSet[ReferenceEqualPlanWrapper]()
+ mWrapper.add(planWrapper)
+ (m, mWrapper, false)
}
case plan: LogicalPlan =>
- val relations = ArrayBuffer.empty[MultiInstanceRelation]
+ val relations = new mutable.HashSet[ReferenceEqualPlanWrapper]()
+ var planChanged = false
val newPlan = if (plan.children.nonEmpty) {
- val newChildren = ArrayBuffer.empty[LogicalPlan]
+ val newChildren = mutable.ArrayBuffer.empty[LogicalPlan]
for (c <- plan.children) {
- val (renewed, collected) = renewDuplicatedRelations(existingRelations ++ relations, c)
+ val (renewed, collected, changed) =
+ renewDuplicatedRelations(existingRelations ++ relations, c)
newChildren += renewed
relations ++= collected
+ if (changed) {
+ planChanged = true
+ }
}
- if (plan.childrenResolved) {
- val attrMap = AttributeMap(
- plan
- .children
- .flatMap(_.output).zip(newChildren.flatMap(_.output))
- .filter { case (a1, a2) => a1.exprId != a2.exprId }
- )
- plan.withNewChildren(newChildren.toSeq).rewriteAttrs(attrMap)
+ if (planChanged) {
+ if (plan.childrenResolved) {
+ val planWithNewChildren = plan.withNewChildren(newChildren.toSeq)
+ val attrMap = AttributeMap(
+ plan
+ .children
+ .flatMap(_.output).zip(newChildren.flatMap(_.output))
+ .filter { case (a1, a2) => a1.exprId != a2.exprId }
+ )
+ if (attrMap.isEmpty) {
+ planWithNewChildren
+ } else {
+ planWithNewChildren.rewriteAttrs(attrMap)
+ }
+ } else {
+ plan.withNewChildren(newChildren.toSeq)
+ }
} else {
- plan.withNewChildren(newChildren.toSeq)
+ plan
}
} else {
plan
}
val planWithNewSubquery = newPlan.transformExpressions {
case subquery: SubqueryExpression =>
- val (renewed, collected) = renewDuplicatedRelations(
+ val (renewed, collected, changed) = renewDuplicatedRelations(
existingRelations ++ relations, subquery.plan)
relations ++= collected
+ if (changed) planChanged = true
subquery.withNewPlan(renewed)
}
- (planWithNewSubquery, relations.toSeq)
+ (planWithNewSubquery, relations, planChanged)
}
+ @inline
private def isDuplicated(
- existingRelations: Seq[MultiInstanceRelation],
- relation: MultiInstanceRelation): Boolean = {
- existingRelations.exists { er =>
- er.asInstanceOf[LogicalPlan].outputSet
- .intersect(relation.asInstanceOf[LogicalPlan].outputSet).nonEmpty
- }
+ existingRelations: mutable.HashSet[ReferenceEqualPlanWrapper],
+ planWrapper: ReferenceEqualPlanWrapper): Boolean = {
+ existingRelations.contains(planWrapper)
Review comment:
So the logic of `isDuplicated` becomes checking whether the plan exists instead of checking outputSet intersection? Is that a safe change?
--
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.
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] SparkQA removed a comment on pull request #32590: [SPARK-35445][SQL] Reduce the execution time of DeduplicateRelations
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #32590:
URL: https://github.com/apache/spark/pull/32590#issuecomment-844180116
**[Test build #138712 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/138712/testReport)** for PR 32590 at commit [`f085918`](https://github.com/apache/spark/commit/f0859185cad086a8a7fe4732cc3ca1de3834bfee).
--
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.
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] Ngone51 commented on a change in pull request #32590: [SPARK-35445][SQL] Reduce the execution time of DeduplicateRelations
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on a change in pull request #32590:
URL: https://github.com/apache/spark/pull/32590#discussion_r635367649
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/DeduplicateRelations.scala
##########
@@ -67,61 +81,78 @@ object DeduplicateRelations extends Rule[LogicalPlan] {
* all relations of the new LogicalPlan )
*/
private def renewDuplicatedRelations(
- existingRelations: Seq[MultiInstanceRelation],
- plan: LogicalPlan): (LogicalPlan, Seq[MultiInstanceRelation]) = plan match {
- case p: LogicalPlan if p.isStreaming => (plan, Nil)
+ existingRelations: mutable.HashSet[ReferenceEqualPlanWrapper],
+ plan: LogicalPlan)
+ : (LogicalPlan, mutable.HashSet[ReferenceEqualPlanWrapper], Boolean) = plan match {
+ case p: LogicalPlan if p.isStreaming => (plan, mutable.HashSet.empty, false)
case m: MultiInstanceRelation =>
- if (isDuplicated(existingRelations, m)) {
+ val planWrapper = ReferenceEqualPlanWrapper(m)
+ if (isDuplicated(existingRelations, planWrapper)) {
val newNode = m.newInstance()
newNode.copyTagsFrom(m)
- (newNode, Nil)
+ (newNode, mutable.HashSet.empty, true)
} else {
- (m, Seq(m))
+ val mWrapper = new mutable.HashSet[ReferenceEqualPlanWrapper]()
+ mWrapper.add(planWrapper)
+ (m, mWrapper, false)
}
case plan: LogicalPlan =>
- val relations = ArrayBuffer.empty[MultiInstanceRelation]
+ val relations = new mutable.HashSet[ReferenceEqualPlanWrapper]()
+ var planChanged = false
val newPlan = if (plan.children.nonEmpty) {
- val newChildren = ArrayBuffer.empty[LogicalPlan]
+ val newChildren = mutable.ArrayBuffer.empty[LogicalPlan]
for (c <- plan.children) {
- val (renewed, collected) = renewDuplicatedRelations(existingRelations ++ relations, c)
+ val (renewed, collected, changed) =
+ renewDuplicatedRelations(existingRelations ++ relations, c)
newChildren += renewed
relations ++= collected
+ if (changed) {
+ planChanged = true
+ }
}
- if (plan.childrenResolved) {
- val attrMap = AttributeMap(
- plan
- .children
- .flatMap(_.output).zip(newChildren.flatMap(_.output))
- .filter { case (a1, a2) => a1.exprId != a2.exprId }
- )
- plan.withNewChildren(newChildren.toSeq).rewriteAttrs(attrMap)
+ if (planChanged) {
+ if (plan.childrenResolved) {
+ val planWithNewChildren = plan.withNewChildren(newChildren.toSeq)
+ val attrMap = AttributeMap(
+ plan
+ .children
+ .flatMap(_.output).zip(newChildren.flatMap(_.output))
+ .filter { case (a1, a2) => a1.exprId != a2.exprId }
+ )
+ if (attrMap.isEmpty) {
+ planWithNewChildren
+ } else {
+ planWithNewChildren.rewriteAttrs(attrMap)
+ }
+ } else {
+ plan.withNewChildren(newChildren.toSeq)
+ }
} else {
- plan.withNewChildren(newChildren.toSeq)
+ plan
}
} else {
plan
}
val planWithNewSubquery = newPlan.transformExpressions {
case subquery: SubqueryExpression =>
- val (renewed, collected) = renewDuplicatedRelations(
+ val (renewed, collected, changed) = renewDuplicatedRelations(
existingRelations ++ relations, subquery.plan)
relations ++= collected
+ if (changed) planChanged = true
subquery.withNewPlan(renewed)
}
- (planWithNewSubquery, relations.toSeq)
+ (planWithNewSubquery, relations, planChanged)
}
+ @inline
private def isDuplicated(
- existingRelations: Seq[MultiInstanceRelation],
- relation: MultiInstanceRelation): Boolean = {
- existingRelations.exists { er =>
- er.asInstanceOf[LogicalPlan].outputSet
- .intersect(relation.asInstanceOf[LogicalPlan].outputSet).nonEmpty
- }
+ existingRelations: mutable.HashSet[ReferenceEqualPlanWrapper],
+ planWrapper: ReferenceEqualPlanWrapper): Boolean = {
+ existingRelations.contains(planWrapper)
Review comment:
It's safe. For the `MultiInstanceRelation`, it's impossible to have partial duplicate attributes with others as its attributes are always generated by itself, which results in the monotonically increasing exprid.
--
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.
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 change in pull request #32590: [SPARK-35445][SQL] Reduce the execution time of DeduplicateRelations
Posted by GitBox <gi...@apache.org>.
cloud-fan commented on a change in pull request #32590:
URL: https://github.com/apache/spark/pull/32590#discussion_r635759716
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/DeduplicateRelations.scala
##########
@@ -67,61 +81,78 @@ object DeduplicateRelations extends Rule[LogicalPlan] {
* all relations of the new LogicalPlan )
*/
private def renewDuplicatedRelations(
- existingRelations: Seq[MultiInstanceRelation],
- plan: LogicalPlan): (LogicalPlan, Seq[MultiInstanceRelation]) = plan match {
- case p: LogicalPlan if p.isStreaming => (plan, Nil)
+ existingRelations: mutable.HashSet[ReferenceEqualPlanWrapper],
+ plan: LogicalPlan)
+ : (LogicalPlan, mutable.HashSet[ReferenceEqualPlanWrapper], Boolean) = plan match {
+ case p: LogicalPlan if p.isStreaming => (plan, mutable.HashSet.empty, false)
case m: MultiInstanceRelation =>
- if (isDuplicated(existingRelations, m)) {
+ val planWrapper = ReferenceEqualPlanWrapper(m)
+ if (isDuplicated(existingRelations, planWrapper)) {
val newNode = m.newInstance()
newNode.copyTagsFrom(m)
- (newNode, Nil)
+ (newNode, mutable.HashSet.empty, true)
} else {
- (m, Seq(m))
+ val mWrapper = new mutable.HashSet[ReferenceEqualPlanWrapper]()
+ mWrapper.add(planWrapper)
+ (m, mWrapper, false)
}
case plan: LogicalPlan =>
- val relations = ArrayBuffer.empty[MultiInstanceRelation]
+ val relations = new mutable.HashSet[ReferenceEqualPlanWrapper]()
+ var planChanged = false
val newPlan = if (plan.children.nonEmpty) {
- val newChildren = ArrayBuffer.empty[LogicalPlan]
+ val newChildren = mutable.ArrayBuffer.empty[LogicalPlan]
for (c <- plan.children) {
- val (renewed, collected) = renewDuplicatedRelations(existingRelations ++ relations, c)
+ val (renewed, collected, changed) =
+ renewDuplicatedRelations(existingRelations ++ relations, c)
newChildren += renewed
relations ++= collected
+ if (changed) {
+ planChanged = true
+ }
}
- if (plan.childrenResolved) {
- val attrMap = AttributeMap(
- plan
- .children
- .flatMap(_.output).zip(newChildren.flatMap(_.output))
- .filter { case (a1, a2) => a1.exprId != a2.exprId }
- )
- plan.withNewChildren(newChildren.toSeq).rewriteAttrs(attrMap)
+ if (planChanged) {
+ if (plan.childrenResolved) {
+ val planWithNewChildren = plan.withNewChildren(newChildren.toSeq)
+ val attrMap = AttributeMap(
+ plan
+ .children
+ .flatMap(_.output).zip(newChildren.flatMap(_.output))
+ .filter { case (a1, a2) => a1.exprId != a2.exprId }
+ )
+ if (attrMap.isEmpty) {
+ planWithNewChildren
+ } else {
+ planWithNewChildren.rewriteAttrs(attrMap)
+ }
+ } else {
+ plan.withNewChildren(newChildren.toSeq)
+ }
} else {
- plan.withNewChildren(newChildren.toSeq)
+ plan
}
} else {
plan
}
val planWithNewSubquery = newPlan.transformExpressions {
case subquery: SubqueryExpression =>
- val (renewed, collected) = renewDuplicatedRelations(
+ val (renewed, collected, changed) = renewDuplicatedRelations(
existingRelations ++ relations, subquery.plan)
relations ++= collected
+ if (changed) planChanged = true
subquery.withNewPlan(renewed)
}
- (planWithNewSubquery, relations.toSeq)
+ (planWithNewSubquery, relations, planChanged)
}
+ @inline
Review comment:
Instead of calling this function, can we manually inline it and write `existingRelations.contains(planWrapper)`?
--
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.
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] SparkQA commented on pull request #32590: [SPARK-35445][SQL] Reduce the execution time of DeduplicateRelations
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32590:
URL: https://github.com/apache/spark/pull/32590#issuecomment-845597633
**[Test build #138782 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/138782/testReport)** for PR 32590 at commit [`47c603d`](https://github.com/apache/spark/commit/47c603d7718bde49729d5b501d75c1b55138a58d).
--
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.
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] SparkQA commented on pull request #32590: [SPARK-35445][SQL] Reduce the execution time of DeduplicateRelations
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32590:
URL: https://github.com/apache/spark/pull/32590#issuecomment-844180116
**[Test build #138712 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/138712/testReport)** for PR 32590 at commit [`f085918`](https://github.com/apache/spark/commit/f0859185cad086a8a7fe4732cc3ca1de3834bfee).
--
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.
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] SparkQA commented on pull request #32590: [SPARK-35445][SQL] Reduce the execution time of DeduplicateRelations
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32590:
URL: https://github.com/apache/spark/pull/32590#issuecomment-844879414
**[Test build #138737 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/138737/testReport)** for PR 32590 at commit [`ef50b77`](https://github.com/apache/spark/commit/ef50b77b580e3b9a2ed57b8bc22258569f596212).
--
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.
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] gengliangwang commented on a change in pull request #32590: [SPARK-35445][SQL] Reduce the execution time of DeduplicateRelations
Posted by GitBox <gi...@apache.org>.
gengliangwang commented on a change in pull request #32590:
URL: https://github.com/apache/spark/pull/32590#discussion_r635352040
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/DeduplicateRelations.scala
##########
@@ -17,16 +17,30 @@
package org.apache.spark.sql.catalyst.analysis
-import scala.collection.mutable.ArrayBuffer
+import scala.collection.mutable
import org.apache.spark.sql.catalyst.expressions.{Alias, AttributeMap, AttributeSet, NamedExpression, SubqueryExpression}
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.trees.AlwaysProcess
+/**
+ * A help class that used to detect duplicate relations fast in `DeduplicateRelations`
+ */
+case class ReferenceEqualPlanWrapper(plan: LogicalPlan) {
+ private val _hashCode = System.identityHashCode(plan)
Review comment:
Why using System.identityHashCode instead of plan.hash()?
--
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.
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] gengliangwang closed pull request #32590: [SPARK-35445][SQL] Reduce the execution time of DeduplicateRelations
Posted by GitBox <gi...@apache.org>.
gengliangwang closed pull request #32590:
URL: https://github.com/apache/spark/pull/32590
--
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.
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] AmplabJenkins removed a comment on pull request #32590: [SPARK-35445][SQL] Reduce the execution time of DeduplicateRelations
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32590:
URL: https://github.com/apache/spark/pull/32590#issuecomment-845144933
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/138737/
--
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.
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] SparkQA removed a comment on pull request #32590: [SPARK-35445][SQL] Reduce the execution time of DeduplicateRelations
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #32590:
URL: https://github.com/apache/spark/pull/32590#issuecomment-844879414
**[Test build #138737 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/138737/testReport)** for PR 32590 at commit [`ef50b77`](https://github.com/apache/spark/commit/ef50b77b580e3b9a2ed57b8bc22258569f596212).
--
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.
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] maropu commented on a change in pull request #32590: [SPARK-35445][SQL] Reduce the execution time of DeduplicateRelations
Posted by GitBox <gi...@apache.org>.
maropu commented on a change in pull request #32590:
URL: https://github.com/apache/spark/pull/32590#discussion_r635670349
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/DeduplicateRelations.scala
##########
@@ -67,61 +81,78 @@ object DeduplicateRelations extends Rule[LogicalPlan] {
* all relations of the new LogicalPlan )
*/
private def renewDuplicatedRelations(
- existingRelations: Seq[MultiInstanceRelation],
- plan: LogicalPlan): (LogicalPlan, Seq[MultiInstanceRelation]) = plan match {
- case p: LogicalPlan if p.isStreaming => (plan, Nil)
+ existingRelations: mutable.HashSet[ReferenceEqualPlanWrapper],
+ plan: LogicalPlan)
+ : (LogicalPlan, mutable.HashSet[ReferenceEqualPlanWrapper], Boolean) = plan match {
+ case p: LogicalPlan if p.isStreaming => (plan, mutable.HashSet.empty, false)
case m: MultiInstanceRelation =>
- if (isDuplicated(existingRelations, m)) {
+ val planWrapper = ReferenceEqualPlanWrapper(m)
+ if (isDuplicated(existingRelations, planWrapper)) {
val newNode = m.newInstance()
newNode.copyTagsFrom(m)
- (newNode, Nil)
+ (newNode, mutable.HashSet.empty, true)
} else {
- (m, Seq(m))
+ val mWrapper = new mutable.HashSet[ReferenceEqualPlanWrapper]()
+ mWrapper.add(planWrapper)
+ (m, mWrapper, false)
}
case plan: LogicalPlan =>
- val relations = ArrayBuffer.empty[MultiInstanceRelation]
+ val relations = new mutable.HashSet[ReferenceEqualPlanWrapper]()
+ var planChanged = false
val newPlan = if (plan.children.nonEmpty) {
- val newChildren = ArrayBuffer.empty[LogicalPlan]
+ val newChildren = mutable.ArrayBuffer.empty[LogicalPlan]
for (c <- plan.children) {
- val (renewed, collected) = renewDuplicatedRelations(existingRelations ++ relations, c)
+ val (renewed, collected, changed) =
+ renewDuplicatedRelations(existingRelations ++ relations, c)
newChildren += renewed
relations ++= collected
+ if (changed) {
+ planChanged = true
+ }
}
- if (plan.childrenResolved) {
- val attrMap = AttributeMap(
- plan
- .children
- .flatMap(_.output).zip(newChildren.flatMap(_.output))
- .filter { case (a1, a2) => a1.exprId != a2.exprId }
- )
- plan.withNewChildren(newChildren.toSeq).rewriteAttrs(attrMap)
+ if (planChanged) {
+ if (plan.childrenResolved) {
+ val planWithNewChildren = plan.withNewChildren(newChildren.toSeq)
+ val attrMap = AttributeMap(
+ plan
+ .children
+ .flatMap(_.output).zip(newChildren.flatMap(_.output))
+ .filter { case (a1, a2) => a1.exprId != a2.exprId }
+ )
+ if (attrMap.isEmpty) {
+ planWithNewChildren
+ } else {
+ planWithNewChildren.rewriteAttrs(attrMap)
+ }
+ } else {
+ plan.withNewChildren(newChildren.toSeq)
+ }
} else {
- plan.withNewChildren(newChildren.toSeq)
+ plan
}
} else {
plan
}
val planWithNewSubquery = newPlan.transformExpressions {
case subquery: SubqueryExpression =>
- val (renewed, collected) = renewDuplicatedRelations(
+ val (renewed, collected, changed) = renewDuplicatedRelations(
existingRelations ++ relations, subquery.plan)
relations ++= collected
+ if (changed) planChanged = true
subquery.withNewPlan(renewed)
}
- (planWithNewSubquery, relations.toSeq)
+ (planWithNewSubquery, relations, planChanged)
}
+ @inline
Review comment:
This annotation can also reduce the running time? (The PR description says nothing though)
--
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.
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] SparkQA commented on pull request #32590: [SPARK-35445][SQL] Reduce the execution time of DeduplicateRelations
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32590:
URL: https://github.com/apache/spark/pull/32590#issuecomment-844958069
Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/43260/
--
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.
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] gengliangwang commented on pull request #32590: [SPARK-35445][SQL] Reduce the execution time of DeduplicateRelations
Posted by GitBox <gi...@apache.org>.
gengliangwang commented on pull request #32590:
URL: https://github.com/apache/spark/pull/32590#issuecomment-845662291
Thanks, merging to master
--
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.
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] SparkQA commented on pull request #32590: [SPARK-35445][SQL] Reduce the execution time of DeduplicateRelations
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32590:
URL: https://github.com/apache/spark/pull/32590#issuecomment-844254505
Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/43233/
--
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.
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] gengliangwang commented on a change in pull request #32590: [SPARK-35445][SQL] Reduce the execution time of DeduplicateRelations
Posted by GitBox <gi...@apache.org>.
gengliangwang commented on a change in pull request #32590:
URL: https://github.com/apache/spark/pull/32590#discussion_r636057267
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/DeduplicateRelations.scala
##########
@@ -64,64 +78,74 @@ object DeduplicateRelations extends Rule[LogicalPlan] {
* @param existingRelations the known unique relations for a LogicalPlan
* @param plan the LogicalPlan that requires the deduplication
* @return (the new LogicalPlan which already deduplicate all duplicated relations (if any),
- * all relations of the new LogicalPlan )
+ * all relations of the new LogicalPlan, whether the plan has changed or not)
Review comment:
nit: `the plan has changed or not` => `the plan is changed or not`
--
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.
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] AmplabJenkins removed a comment on pull request #32590: [SPARK-35445][SQL] Reduce the execution time of DeduplicateRelations
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32590:
URL: https://github.com/apache/spark/pull/32590#issuecomment-845708887
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/43299/
--
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.
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] Ngone51 commented on pull request #32590: [SPARK-35445][SQL] Reduce the execution time of DeduplicateRelations
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on pull request #32590:
URL: https://github.com/apache/spark/pull/32590#issuecomment-844144147
cc @cloud-fan @maropu Please take a look, thanks!
--
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.
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] Ngone51 commented on a change in pull request #32590: [SPARK-35445][SQL] Reduce the execution time of DeduplicateRelations
Posted by GitBox <gi...@apache.org>.
Ngone51 commented on a change in pull request #32590:
URL: https://github.com/apache/spark/pull/32590#discussion_r635896472
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/DeduplicateRelations.scala
##########
@@ -67,61 +81,78 @@ object DeduplicateRelations extends Rule[LogicalPlan] {
* all relations of the new LogicalPlan )
*/
private def renewDuplicatedRelations(
- existingRelations: Seq[MultiInstanceRelation],
- plan: LogicalPlan): (LogicalPlan, Seq[MultiInstanceRelation]) = plan match {
- case p: LogicalPlan if p.isStreaming => (plan, Nil)
+ existingRelations: mutable.HashSet[ReferenceEqualPlanWrapper],
+ plan: LogicalPlan)
+ : (LogicalPlan, mutable.HashSet[ReferenceEqualPlanWrapper], Boolean) = plan match {
+ case p: LogicalPlan if p.isStreaming => (plan, mutable.HashSet.empty, false)
case m: MultiInstanceRelation =>
- if (isDuplicated(existingRelations, m)) {
+ val planWrapper = ReferenceEqualPlanWrapper(m)
+ if (isDuplicated(existingRelations, planWrapper)) {
val newNode = m.newInstance()
newNode.copyTagsFrom(m)
- (newNode, Nil)
+ (newNode, mutable.HashSet.empty, true)
} else {
- (m, Seq(m))
+ val mWrapper = new mutable.HashSet[ReferenceEqualPlanWrapper]()
+ mWrapper.add(planWrapper)
+ (m, mWrapper, false)
}
case plan: LogicalPlan =>
- val relations = ArrayBuffer.empty[MultiInstanceRelation]
+ val relations = new mutable.HashSet[ReferenceEqualPlanWrapper]()
+ var planChanged = false
val newPlan = if (plan.children.nonEmpty) {
- val newChildren = ArrayBuffer.empty[LogicalPlan]
+ val newChildren = mutable.ArrayBuffer.empty[LogicalPlan]
for (c <- plan.children) {
- val (renewed, collected) = renewDuplicatedRelations(existingRelations ++ relations, c)
+ val (renewed, collected, changed) =
+ renewDuplicatedRelations(existingRelations ++ relations, c)
newChildren += renewed
relations ++= collected
+ if (changed) {
+ planChanged = true
+ }
}
- if (plan.childrenResolved) {
- val attrMap = AttributeMap(
- plan
- .children
- .flatMap(_.output).zip(newChildren.flatMap(_.output))
- .filter { case (a1, a2) => a1.exprId != a2.exprId }
- )
- plan.withNewChildren(newChildren.toSeq).rewriteAttrs(attrMap)
+ if (planChanged) {
+ if (plan.childrenResolved) {
+ val planWithNewChildren = plan.withNewChildren(newChildren.toSeq)
+ val attrMap = AttributeMap(
+ plan
+ .children
+ .flatMap(_.output).zip(newChildren.flatMap(_.output))
+ .filter { case (a1, a2) => a1.exprId != a2.exprId }
+ )
+ if (attrMap.isEmpty) {
+ planWithNewChildren
+ } else {
+ planWithNewChildren.rewriteAttrs(attrMap)
+ }
+ } else {
+ plan.withNewChildren(newChildren.toSeq)
+ }
} else {
- plan.withNewChildren(newChildren.toSeq)
+ plan
}
} else {
plan
}
val planWithNewSubquery = newPlan.transformExpressions {
case subquery: SubqueryExpression =>
- val (renewed, collected) = renewDuplicatedRelations(
+ val (renewed, collected, changed) = renewDuplicatedRelations(
existingRelations ++ relations, subquery.plan)
relations ++= collected
+ if (changed) planChanged = true
subquery.withNewPlan(renewed)
}
- (planWithNewSubquery, relations.toSeq)
+ (planWithNewSubquery, relations, planChanged)
}
+ @inline
Review comment:
> This annotation can also reduce the running time? (The PR description says nothing though)
Not really. It's actually an experienced change.
--
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.
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] AmplabJenkins commented on pull request #32590: [SPARK-35445][SQL] Reduce the execution time of DeduplicateRelations
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32590:
URL: https://github.com/apache/spark/pull/32590#issuecomment-845708887
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/43299/
--
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.
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] maropu commented on a change in pull request #32590: [SPARK-35445][SQL] Reduce the execution time of DeduplicateRelations
Posted by GitBox <gi...@apache.org>.
maropu commented on a change in pull request #32590:
URL: https://github.com/apache/spark/pull/32590#discussion_r635668680
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/DeduplicateRelations.scala
##########
@@ -17,16 +17,30 @@
package org.apache.spark.sql.catalyst.analysis
-import scala.collection.mutable.ArrayBuffer
+import scala.collection.mutable
import org.apache.spark.sql.catalyst.expressions.{Alias, AttributeMap, AttributeSet, NamedExpression, SubqueryExpression}
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.trees.AlwaysProcess
+/**
+ * A help class that used to detect duplicate relations fast in `DeduplicateRelations`
Review comment:
nit: That might be, `A helper class used to...`
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/DeduplicateRelations.scala
##########
@@ -67,61 +81,78 @@ object DeduplicateRelations extends Rule[LogicalPlan] {
* all relations of the new LogicalPlan )
*/
private def renewDuplicatedRelations(
- existingRelations: Seq[MultiInstanceRelation],
- plan: LogicalPlan): (LogicalPlan, Seq[MultiInstanceRelation]) = plan match {
- case p: LogicalPlan if p.isStreaming => (plan, Nil)
+ existingRelations: mutable.HashSet[ReferenceEqualPlanWrapper],
+ plan: LogicalPlan)
+ : (LogicalPlan, mutable.HashSet[ReferenceEqualPlanWrapper], Boolean) = plan match {
+ case p: LogicalPlan if p.isStreaming => (plan, mutable.HashSet.empty, false)
case m: MultiInstanceRelation =>
- if (isDuplicated(existingRelations, m)) {
+ val planWrapper = ReferenceEqualPlanWrapper(m)
+ if (isDuplicated(existingRelations, planWrapper)) {
val newNode = m.newInstance()
newNode.copyTagsFrom(m)
- (newNode, Nil)
+ (newNode, mutable.HashSet.empty, true)
} else {
- (m, Seq(m))
+ val mWrapper = new mutable.HashSet[ReferenceEqualPlanWrapper]()
+ mWrapper.add(planWrapper)
+ (m, mWrapper, false)
}
case plan: LogicalPlan =>
- val relations = ArrayBuffer.empty[MultiInstanceRelation]
+ val relations = new mutable.HashSet[ReferenceEqualPlanWrapper]()
+ var planChanged = false
val newPlan = if (plan.children.nonEmpty) {
- val newChildren = ArrayBuffer.empty[LogicalPlan]
+ val newChildren = mutable.ArrayBuffer.empty[LogicalPlan]
for (c <- plan.children) {
- val (renewed, collected) = renewDuplicatedRelations(existingRelations ++ relations, c)
+ val (renewed, collected, changed) =
+ renewDuplicatedRelations(existingRelations ++ relations, c)
newChildren += renewed
relations ++= collected
+ if (changed) {
+ planChanged = true
+ }
}
- if (plan.childrenResolved) {
- val attrMap = AttributeMap(
- plan
- .children
- .flatMap(_.output).zip(newChildren.flatMap(_.output))
- .filter { case (a1, a2) => a1.exprId != a2.exprId }
- )
- plan.withNewChildren(newChildren.toSeq).rewriteAttrs(attrMap)
+ if (planChanged) {
+ if (plan.childrenResolved) {
+ val planWithNewChildren = plan.withNewChildren(newChildren.toSeq)
+ val attrMap = AttributeMap(
+ plan
+ .children
+ .flatMap(_.output).zip(newChildren.flatMap(_.output))
+ .filter { case (a1, a2) => a1.exprId != a2.exprId }
+ )
+ if (attrMap.isEmpty) {
+ planWithNewChildren
+ } else {
+ planWithNewChildren.rewriteAttrs(attrMap)
+ }
+ } else {
+ plan.withNewChildren(newChildren.toSeq)
+ }
} else {
- plan.withNewChildren(newChildren.toSeq)
+ plan
}
} else {
plan
}
val planWithNewSubquery = newPlan.transformExpressions {
case subquery: SubqueryExpression =>
- val (renewed, collected) = renewDuplicatedRelations(
+ val (renewed, collected, changed) = renewDuplicatedRelations(
existingRelations ++ relations, subquery.plan)
relations ++= collected
+ if (changed) planChanged = true
subquery.withNewPlan(renewed)
}
- (planWithNewSubquery, relations.toSeq)
+ (planWithNewSubquery, relations, planChanged)
}
+ @inline
Review comment:
This annotation can also reduce the running time? (The PR description say nothing though)
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/DeduplicateRelations.scala
##########
@@ -17,16 +17,30 @@
package org.apache.spark.sql.catalyst.analysis
-import scala.collection.mutable.ArrayBuffer
+import scala.collection.mutable
import org.apache.spark.sql.catalyst.expressions.{Alias, AttributeMap, AttributeSet, NamedExpression, SubqueryExpression}
import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.trees.AlwaysProcess
+/**
+ * A help class that used to detect duplicate relations fast in `DeduplicateRelations`
+ */
+case class ReferenceEqualPlanWrapper(plan: LogicalPlan) {
+ private val _hashCode = System.identityHashCode(plan)
Review comment:
Ah, I see. The java library does't have `IdentityHasSet`, even though it has `IdentityHasMap`...
--
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.
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] AmplabJenkins removed a comment on pull request #32590: [SPARK-35445][SQL] Reduce the execution time of DeduplicateRelations
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32590:
URL: https://github.com/apache/spark/pull/32590#issuecomment-845867966
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/138782/
--
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.
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] gengliangwang commented on a change in pull request #32590: [SPARK-35445][SQL] Reduce the execution time of DeduplicateRelations
Posted by GitBox <gi...@apache.org>.
gengliangwang commented on a change in pull request #32590:
URL: https://github.com/apache/spark/pull/32590#discussion_r635340530
##########
File path: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/DeduplicateRelations.scala
##########
@@ -67,61 +81,78 @@ object DeduplicateRelations extends Rule[LogicalPlan] {
* all relations of the new LogicalPlan )
*/
private def renewDuplicatedRelations(
- existingRelations: Seq[MultiInstanceRelation],
- plan: LogicalPlan): (LogicalPlan, Seq[MultiInstanceRelation]) = plan match {
- case p: LogicalPlan if p.isStreaming => (plan, Nil)
+ existingRelations: mutable.HashSet[ReferenceEqualPlanWrapper],
+ plan: LogicalPlan)
+ : (LogicalPlan, mutable.HashSet[ReferenceEqualPlanWrapper], Boolean) = plan match {
Review comment:
add doc for the returned `Boolean` value
--
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.
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