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 <gi...@git.apache.org> on 2017/12/27 18:57:50 UTC
[GitHub] spark pull request #20094: [SPARK-20392][SQL][followup] should not add extra...
GitHub user cloud-fan opened a pull request:
https://github.com/apache/spark/pull/20094
[SPARK-20392][SQL][followup] should not add extra AnalysisBarrier
## What changes were proposed in this pull request?
I found this problem while auditing the analyzer code. It's dangerous to introduce extra `AnalysisBarrer` during analysis, as the plan inside it will bypass all analysis afterward, which may not be expected.
## How was this patch tested?
existing tests
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/cloud-fan/spark barrier
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/20094.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #20094
----
commit 64709fc78ad49d2dd260d8f3a7a96d10f0081cbb
Author: Wenchen Fan <we...@...>
Date: 2017-12-27T18:54:50Z
should not add extra AnalysisBarrier
----
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20094: [SPARK-20392][SQL][followup] should not add extra Analys...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/20094
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/85453/
Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20094: [SPARK-20392][SQL][followup] should not add extra Analys...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/20094
**[Test build #85439 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/85439/testReport)** for PR 20094 at commit [`64709fc`](https://github.com/apache/spark/commit/64709fc78ad49d2dd260d8f3a7a96d10f0081cbb).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #20094: [SPARK-20392][SQL][followup] should not add extra...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/20094#discussion_r158890342
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala ---
@@ -1079,100 +1083,76 @@ class Analyzer(
case sa @ Sort(_, _, AnalysisBarrier(child: Aggregate)) => sa
case sa @ Sort(_, _, child: Aggregate) => sa
- case s @ Sort(order, _, originalChild) if !s.resolved && originalChild.resolved =>
- val child = EliminateBarriers(originalChild)
- try {
- val newOrder = order.map(resolveExpressionRecursively(_, child).asInstanceOf[SortOrder])
- val requiredAttrs = AttributeSet(newOrder).filter(_.resolved)
- val missingAttrs = requiredAttrs -- child.outputSet
- if (missingAttrs.nonEmpty) {
- // Add missing attributes and then project them away after the sort.
- Project(child.output,
- Sort(newOrder, s.global, addMissingAttr(child, missingAttrs)))
- } else if (newOrder != order) {
- s.copy(order = newOrder)
- } else {
- s
- }
- } catch {
- // Attempting to resolve it might fail. When this happens, return the original plan.
- // Users will see an AnalysisException for resolution failure of missing attributes
- // in Sort
- case ae: AnalysisException => s
+ case s @ Sort(order, _, child) if !s.resolved && child.resolved =>
+ val (newOrder, newChild) = resolveExprsAndAddMissingAttrs(order, child)
+ val ordering = newOrder.map(_.asInstanceOf[SortOrder])
+ if (child.output == newChild.output) {
+ s.copy(order = ordering)
+ } else {
+ // Add missing attributes and then project them away.
+ val newSort = s.copy(order = ordering, child = newChild)
+ Project(child.output, newSort)
}
- case f @ Filter(cond, originalChild) if !f.resolved && originalChild.resolved =>
- val child = EliminateBarriers(originalChild)
- try {
- val newCond = resolveExpressionRecursively(cond, child)
- val requiredAttrs = newCond.references.filter(_.resolved)
- val missingAttrs = requiredAttrs -- child.outputSet
- if (missingAttrs.nonEmpty) {
- // Add missing attributes and then project them away.
- Project(child.output,
- Filter(newCond, addMissingAttr(child, missingAttrs)))
- } else if (newCond != cond) {
- f.copy(condition = newCond)
- } else {
- f
- }
- } catch {
- // Attempting to resolve it might fail. When this happens, return the original plan.
- // Users will see an AnalysisException for resolution failure of missing attributes
- case ae: AnalysisException => f
+ case f @ Filter(cond, child) if !f.resolved && child.resolved =>
+ val (newCond, newChild) = resolveExprsAndAddMissingAttrs(Seq(cond), child)
+ if (child.output == newChild.output) {
+ f.copy(condition = newCond.head)
+ } else {
+ // Add missing attributes and then project them away.
+ val newFilter = Filter(newCond.head, newChild)
+ Project(child.output, newFilter)
}
}
- /**
- * Add the missing attributes into projectList of Project/Window or aggregateExpressions of
- * Aggregate.
- */
- private def addMissingAttr(plan: LogicalPlan, missingAttrs: AttributeSet): LogicalPlan = {
- if (missingAttrs.isEmpty) {
- return AnalysisBarrier(plan)
- }
- plan match {
- case p: Project =>
- val missing = missingAttrs -- p.child.outputSet
- Project(p.projectList ++ missingAttrs, addMissingAttr(p.child, missing))
- case a: Aggregate =>
- // all the missing attributes should be grouping expressions
- // TODO: push down AggregateExpression
- missingAttrs.foreach { attr =>
- if (!a.groupingExpressions.exists(_.semanticEquals(attr))) {
- throw new AnalysisException(s"Can't add $attr to ${a.simpleString}")
- }
- }
- val newAggregateExpressions = a.aggregateExpressions ++ missingAttrs
- a.copy(aggregateExpressions = newAggregateExpressions)
- case g: Generate =>
- // If join is false, we will convert it to true for getting from the child the missing
- // attributes that its child might have or could have.
- val missing = missingAttrs -- g.child.outputSet
- g.copy(join = true, child = addMissingAttr(g.child, missing))
- case d: Distinct =>
- throw new AnalysisException(s"Can't add $missingAttrs to $d")
- case u: UnaryNode =>
- u.withNewChildren(addMissingAttr(u.child, missingAttrs) :: Nil)
- case other =>
- throw new AnalysisException(s"Can't add $missingAttrs to $other")
- }
- }
-
- /**
- * Resolve the expression on a specified logical plan and it's child (recursively), until
- * the expression is resolved or meet a non-unary node or Subquery.
- */
- @tailrec
- private def resolveExpressionRecursively(expr: Expression, plan: LogicalPlan): Expression = {
- val resolved = resolveExpression(expr, plan)
- if (resolved.resolved) {
- resolved
+ private def resolveExprsAndAddMissingAttrs(
+ exprs: Seq[Expression], plan: LogicalPlan): (Seq[Expression], LogicalPlan) = {
+ if (exprs.forall(_.resolved)) {
+ // All given expressions are resolved, no need to continue anymore.
+ (exprs, plan)
} else {
plan match {
- case u: UnaryNode if !u.isInstanceOf[SubqueryAlias] =>
- resolveExpressionRecursively(resolved, u.child)
- case other => resolved
+ // For `AnalysisBarrier`, recursively resolve expressions and add missing attributes via
+ // its child.
+ case barrier: AnalysisBarrier =>
+ val (newExprs, newChild) = resolveExprsAndAddMissingAttrs(exprs, barrier.child)
+ (newExprs, AnalysisBarrier(newChild))
+
+ case p: Project =>
+ val maybeResolvedExprs = exprs.map(resolveExpression(_, p))
+ val (newExprs, newChild) = resolveExprsAndAddMissingAttrs(maybeResolvedExprs, p.child)
+ val missingAttrs = AttributeSet(newExprs) -- AttributeSet(maybeResolvedExprs)
+ (newExprs, Project(p.projectList ++ missingAttrs, newChild))
+
+ case a @ Aggregate(groupExprs, aggExprs, child) =>
+ val maybeResolvedExprs = exprs.map(resolveExpression(_, a))
+ val (newExprs, newChild) = resolveExprsAndAddMissingAttrs(maybeResolvedExprs, child)
+ val missingAttrs = AttributeSet(newExprs) -- AttributeSet(maybeResolvedExprs)
+ if (missingAttrs.forall(attr => groupExprs.exists(_.semanticEquals(attr)))) {
+ // All the missing attributes are grouping expressions, valid case.
+ (newExprs, a.copy(aggregateExpressions = aggExprs ++ missingAttrs, child = newChild))
+ } else {
+ // Need to add non-grouping attributes, invalid case.
+ (exprs, a)
+ }
+
+ case g: Generate =>
+ val maybeResolvedExprs = exprs.map(resolveExpression(_, g))
+ val (newExprs, newChild) = resolveExprsAndAddMissingAttrs(maybeResolvedExprs, g.child)
+ (newExprs, g.copy(join = true, child = newChild))
+
+ // For `Distinct`, we can't recursively resolve and add attributes via its children.
+ case d: Distinct =>
+ (exprs.map(resolveExpression(_, d)), d)
+
+ case u: UnaryNode =>
--- End diff --
Shouldn't we stop at `SubqueryAlias` as before?
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20094: [SPARK-20392][SQL][followup] should not add extra Analys...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on the issue:
https://github.com/apache/spark/pull/20094
Thanks! Merged to master.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20094: [SPARK-20392][SQL][followup] should not add extra Analys...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/20094
Merged build finished. Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #20094: [SPARK-20392][SQL][followup] should not add extra...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/20094#discussion_r158855682
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala ---
@@ -1404,18 +1384,16 @@ class Analyzer(
*/
object ResolveAggregateFunctions extends Rule[LogicalPlan] {
def apply(plan: LogicalPlan): LogicalPlan = plan.transformUp {
- case filter @ Filter(havingCondition, AnalysisBarrier(aggregate: Aggregate)) =>
- apply(Filter(havingCondition, aggregate)).mapChildren(AnalysisBarrier)
- case filter @ Filter(havingCondition,
- aggregate @ Aggregate(grouping, originalAggExprs, child))
- if aggregate.resolved =>
+ case Filter(cond, AnalysisBarrier(agg: Aggregate)) =>
+ apply(Filter(cond, agg)).mapChildren(AnalysisBarrier)
+ case f @ Filter(cond, agg @ Aggregate(grouping, originalAggExprs, child)) if agg.resolved =>
--- End diff --
just make the names shorter
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20094: [SPARK-20392][SQL][followup] should not add extra Analys...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/20094
Merged build finished. Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20094: [SPARK-20392][SQL][followup] should not add extra Analys...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/20094
Merged build finished. Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20094: [SPARK-20392][SQL][followup] should not add extra Analys...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/20094
**[Test build #85450 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/85450/testReport)** for PR 20094 at commit [`cd39760`](https://github.com/apache/spark/commit/cd397605eaf81e3725396c715602734596a83ac3).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20094: [SPARK-20392][SQL][followup] should not add extra Analys...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/20094
**[Test build #85452 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/85452/testReport)** for PR 20094 at commit [`8879870`](https://github.com/apache/spark/commit/887987015f6ec35cc0e25648f9f714e4b6bfa982).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20094: [SPARK-20392][SQL][followup] should not add extra Analys...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/20094
**[Test build #85452 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/85452/testReport)** for PR 20094 at commit [`8879870`](https://github.com/apache/spark/commit/887987015f6ec35cc0e25648f9f714e4b6bfa982).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20094: [SPARK-20392][SQL][followup] should not add extra Analys...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/20094
**[Test build #85453 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/85453/testReport)** for PR 20094 at commit [`6a25d60`](https://github.com/apache/spark/commit/6a25d60a0e0a24194c1764e217d74d807056039c).
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20094: [SPARK-20392][SQL][followup] should not add extra Analys...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/20094
Merged build finished. Test PASSed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20094: [SPARK-20392][SQL][followup] should not add extra Analys...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/20094
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/85439/
Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20094: [SPARK-20392][SQL][followup] should not add extra Analys...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/20094
**[Test build #85453 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/85453/testReport)** for PR 20094 at commit [`6a25d60`](https://github.com/apache/spark/commit/6a25d60a0e0a24194c1764e217d74d807056039c).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20094: [SPARK-20392][SQL][followup] should not add extra Analys...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on the issue:
https://github.com/apache/spark/pull/20094
LGTM with two minor comments.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #20094: [SPARK-20392][SQL][followup] should not add extra...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/20094#discussion_r158889013
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala ---
@@ -723,7 +726,7 @@ class Analyzer(
s.withNewPlan(dedupOuterReferencesInSubquery(s.plan, attributeRewrites))
}
}
- AnalysisBarrier(newRight)
+ newRight
--- End diff --
newRight is introduced before to be wrapped in `AnalysisBarrier`. We can get rid of it now.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20094: [SPARK-20392][SQL][followup] should not add extra Analys...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on the issue:
https://github.com/apache/spark/pull/20094
cc @viirya @gatorsmile
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #20094: [SPARK-20392][SQL][followup] should not add extra...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/20094#discussion_r158879283
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala ---
@@ -665,14 +664,18 @@ class Analyzer(
* Generate a new logical plan for the right child with different expression IDs
* for all conflicting attributes.
*/
- private def dedupRight (left: LogicalPlan, originalRight: LogicalPlan): LogicalPlan = {
- // Remove analysis barrier if any.
- val right = EliminateBarriers(originalRight)
--- End diff --
If right plan is wrapped (e.g., we join two datasets) in an analysis barrier, the later `right.collect` doesn't work.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20094: [SPARK-20392][SQL][followup] should not add extra Analys...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/20094
**[Test build #85439 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/85439/testReport)** for PR 20094 at commit [`64709fc`](https://github.com/apache/spark/commit/64709fc78ad49d2dd260d8f3a7a96d10f0081cbb).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #20094: [SPARK-20392][SQL][followup] should not add extra...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/20094#discussion_r158891321
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala ---
@@ -1079,100 +1083,76 @@ class Analyzer(
case sa @ Sort(_, _, AnalysisBarrier(child: Aggregate)) => sa
case sa @ Sort(_, _, child: Aggregate) => sa
- case s @ Sort(order, _, originalChild) if !s.resolved && originalChild.resolved =>
- val child = EliminateBarriers(originalChild)
- try {
- val newOrder = order.map(resolveExpressionRecursively(_, child).asInstanceOf[SortOrder])
- val requiredAttrs = AttributeSet(newOrder).filter(_.resolved)
- val missingAttrs = requiredAttrs -- child.outputSet
- if (missingAttrs.nonEmpty) {
- // Add missing attributes and then project them away after the sort.
- Project(child.output,
- Sort(newOrder, s.global, addMissingAttr(child, missingAttrs)))
- } else if (newOrder != order) {
- s.copy(order = newOrder)
- } else {
- s
- }
- } catch {
- // Attempting to resolve it might fail. When this happens, return the original plan.
- // Users will see an AnalysisException for resolution failure of missing attributes
- // in Sort
- case ae: AnalysisException => s
+ case s @ Sort(order, _, child) if !s.resolved && child.resolved =>
+ val (newOrder, newChild) = resolveExprsAndAddMissingAttrs(order, child)
+ val ordering = newOrder.map(_.asInstanceOf[SortOrder])
+ if (child.output == newChild.output) {
+ s.copy(order = ordering)
+ } else {
+ // Add missing attributes and then project them away.
+ val newSort = s.copy(order = ordering, child = newChild)
+ Project(child.output, newSort)
}
- case f @ Filter(cond, originalChild) if !f.resolved && originalChild.resolved =>
- val child = EliminateBarriers(originalChild)
- try {
- val newCond = resolveExpressionRecursively(cond, child)
- val requiredAttrs = newCond.references.filter(_.resolved)
- val missingAttrs = requiredAttrs -- child.outputSet
- if (missingAttrs.nonEmpty) {
- // Add missing attributes and then project them away.
- Project(child.output,
- Filter(newCond, addMissingAttr(child, missingAttrs)))
- } else if (newCond != cond) {
- f.copy(condition = newCond)
- } else {
- f
- }
- } catch {
- // Attempting to resolve it might fail. When this happens, return the original plan.
- // Users will see an AnalysisException for resolution failure of missing attributes
- case ae: AnalysisException => f
+ case f @ Filter(cond, child) if !f.resolved && child.resolved =>
+ val (newCond, newChild) = resolveExprsAndAddMissingAttrs(Seq(cond), child)
+ if (child.output == newChild.output) {
+ f.copy(condition = newCond.head)
+ } else {
+ // Add missing attributes and then project them away.
+ val newFilter = Filter(newCond.head, newChild)
+ Project(child.output, newFilter)
}
}
- /**
- * Add the missing attributes into projectList of Project/Window or aggregateExpressions of
- * Aggregate.
- */
- private def addMissingAttr(plan: LogicalPlan, missingAttrs: AttributeSet): LogicalPlan = {
- if (missingAttrs.isEmpty) {
- return AnalysisBarrier(plan)
- }
- plan match {
- case p: Project =>
- val missing = missingAttrs -- p.child.outputSet
- Project(p.projectList ++ missingAttrs, addMissingAttr(p.child, missing))
- case a: Aggregate =>
- // all the missing attributes should be grouping expressions
- // TODO: push down AggregateExpression
- missingAttrs.foreach { attr =>
- if (!a.groupingExpressions.exists(_.semanticEquals(attr))) {
- throw new AnalysisException(s"Can't add $attr to ${a.simpleString}")
- }
- }
- val newAggregateExpressions = a.aggregateExpressions ++ missingAttrs
- a.copy(aggregateExpressions = newAggregateExpressions)
- case g: Generate =>
- // If join is false, we will convert it to true for getting from the child the missing
- // attributes that its child might have or could have.
- val missing = missingAttrs -- g.child.outputSet
- g.copy(join = true, child = addMissingAttr(g.child, missing))
- case d: Distinct =>
- throw new AnalysisException(s"Can't add $missingAttrs to $d")
- case u: UnaryNode =>
- u.withNewChildren(addMissingAttr(u.child, missingAttrs) :: Nil)
- case other =>
- throw new AnalysisException(s"Can't add $missingAttrs to $other")
- }
- }
-
- /**
- * Resolve the expression on a specified logical plan and it's child (recursively), until
- * the expression is resolved or meet a non-unary node or Subquery.
- */
- @tailrec
- private def resolveExpressionRecursively(expr: Expression, plan: LogicalPlan): Expression = {
- val resolved = resolveExpression(expr, plan)
- if (resolved.resolved) {
- resolved
+ private def resolveExprsAndAddMissingAttrs(
+ exprs: Seq[Expression], plan: LogicalPlan): (Seq[Expression], LogicalPlan) = {
+ if (exprs.forall(_.resolved)) {
+ // All given expressions are resolved, no need to continue anymore.
+ (exprs, plan)
} else {
plan match {
- case u: UnaryNode if !u.isInstanceOf[SubqueryAlias] =>
- resolveExpressionRecursively(resolved, u.child)
- case other => resolved
+ // For `AnalysisBarrier`, recursively resolve expressions and add missing attributes via
+ // its child.
+ case barrier: AnalysisBarrier =>
+ val (newExprs, newChild) = resolveExprsAndAddMissingAttrs(exprs, barrier.child)
+ (newExprs, AnalysisBarrier(newChild))
+
+ case p: Project =>
+ val maybeResolvedExprs = exprs.map(resolveExpression(_, p))
+ val (newExprs, newChild) = resolveExprsAndAddMissingAttrs(maybeResolvedExprs, p.child)
+ val missingAttrs = AttributeSet(newExprs) -- AttributeSet(maybeResolvedExprs)
+ (newExprs, Project(p.projectList ++ missingAttrs, newChild))
+
+ case a @ Aggregate(groupExprs, aggExprs, child) =>
+ val maybeResolvedExprs = exprs.map(resolveExpression(_, a))
+ val (newExprs, newChild) = resolveExprsAndAddMissingAttrs(maybeResolvedExprs, child)
+ val missingAttrs = AttributeSet(newExprs) -- AttributeSet(maybeResolvedExprs)
+ if (missingAttrs.forall(attr => groupExprs.exists(_.semanticEquals(attr)))) {
+ // All the missing attributes are grouping expressions, valid case.
+ (newExprs, a.copy(aggregateExpressions = aggExprs ++ missingAttrs, child = newChild))
+ } else {
+ // Need to add non-grouping attributes, invalid case.
+ (exprs, a)
+ }
+
+ case g: Generate =>
+ val maybeResolvedExprs = exprs.map(resolveExpression(_, g))
+ val (newExprs, newChild) = resolveExprsAndAddMissingAttrs(maybeResolvedExprs, g.child)
+ (newExprs, g.copy(join = true, child = newChild))
+
+ // For `Distinct`, we can't recursively resolve and add attributes via its children.
+ case d: Distinct =>
+ (exprs.map(resolveExpression(_, d)), d)
+
+ case u: UnaryNode =>
--- End diff --
ah good catch! I missed that because the logic was in `resolveExpressionRecursively` instead of `addMissingAttr`.
It indicates that it's more clear to merge these 2 methods :)
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20094: [SPARK-20392][SQL][followup] should not add extra Analys...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/20094
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/85450/
Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #20094: [SPARK-20392][SQL][followup] should not add extra...
Posted by viirya <gi...@git.apache.org>.
Github user viirya commented on a diff in the pull request:
https://github.com/apache/spark/pull/20094#discussion_r158879336
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala ---
@@ -665,14 +664,18 @@ class Analyzer(
* Generate a new logical plan for the right child with different expression IDs
* for all conflicting attributes.
*/
- private def dedupRight (left: LogicalPlan, originalRight: LogicalPlan): LogicalPlan = {
- // Remove analysis barrier if any.
- val right = EliminateBarriers(originalRight)
--- End diff --
oh, I see, you have recursively `dedupRight` on it.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #20094: [SPARK-20392][SQL][followup] should not add extra...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/spark/pull/20094
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20094: [SPARK-20392][SQL][followup] should not add extra Analys...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the issue:
https://github.com/apache/spark/pull/20094
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/85452/
Test FAILed.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request #20094: [SPARK-20392][SQL][followup] should not add extra...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/20094#discussion_r158855737
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala ---
@@ -1079,100 +1083,76 @@ class Analyzer(
case sa @ Sort(_, _, AnalysisBarrier(child: Aggregate)) => sa
case sa @ Sort(_, _, child: Aggregate) => sa
- case s @ Sort(order, _, originalChild) if !s.resolved && originalChild.resolved =>
- val child = EliminateBarriers(originalChild)
- try {
- val newOrder = order.map(resolveExpressionRecursively(_, child).asInstanceOf[SortOrder])
- val requiredAttrs = AttributeSet(newOrder).filter(_.resolved)
- val missingAttrs = requiredAttrs -- child.outputSet
- if (missingAttrs.nonEmpty) {
- // Add missing attributes and then project them away after the sort.
- Project(child.output,
- Sort(newOrder, s.global, addMissingAttr(child, missingAttrs)))
- } else if (newOrder != order) {
- s.copy(order = newOrder)
- } else {
- s
- }
- } catch {
- // Attempting to resolve it might fail. When this happens, return the original plan.
- // Users will see an AnalysisException for resolution failure of missing attributes
- // in Sort
- case ae: AnalysisException => s
+ case s @ Sort(order, _, child) if !s.resolved && child.resolved =>
+ val (newOrder, newChild) = resolveExprsAndAddMissingAttrs(order, child)
+ val ordering = newOrder.map(_.asInstanceOf[SortOrder])
+ if (child.output == newChild.output) {
+ s.copy(order = ordering)
+ } else {
+ // Add missing attributes and then project them away.
+ val newSort = s.copy(order = ordering, child = newChild)
+ Project(child.output, newSort)
}
- case f @ Filter(cond, originalChild) if !f.resolved && originalChild.resolved =>
- val child = EliminateBarriers(originalChild)
- try {
- val newCond = resolveExpressionRecursively(cond, child)
- val requiredAttrs = newCond.references.filter(_.resolved)
- val missingAttrs = requiredAttrs -- child.outputSet
- if (missingAttrs.nonEmpty) {
- // Add missing attributes and then project them away.
- Project(child.output,
- Filter(newCond, addMissingAttr(child, missingAttrs)))
- } else if (newCond != cond) {
- f.copy(condition = newCond)
- } else {
- f
- }
- } catch {
- // Attempting to resolve it might fail. When this happens, return the original plan.
- // Users will see an AnalysisException for resolution failure of missing attributes
- case ae: AnalysisException => f
+ case f @ Filter(cond, child) if !f.resolved && child.resolved =>
+ val (newCond, newChild) = resolveExprsAndAddMissingAttrs(Seq(cond), child)
+ if (child.output == newChild.output) {
+ f.copy(condition = newCond.head)
+ } else {
+ // Add missing attributes and then project them away.
+ val newFilter = Filter(newCond.head, newChild)
+ Project(child.output, newFilter)
}
}
- /**
- * Add the missing attributes into projectList of Project/Window or aggregateExpressions of
- * Aggregate.
- */
- private def addMissingAttr(plan: LogicalPlan, missingAttrs: AttributeSet): LogicalPlan = {
- if (missingAttrs.isEmpty) {
- return AnalysisBarrier(plan)
- }
- plan match {
- case p: Project =>
- val missing = missingAttrs -- p.child.outputSet
- Project(p.projectList ++ missingAttrs, addMissingAttr(p.child, missing))
- case a: Aggregate =>
- // all the missing attributes should be grouping expressions
- // TODO: push down AggregateExpression
- missingAttrs.foreach { attr =>
- if (!a.groupingExpressions.exists(_.semanticEquals(attr))) {
- throw new AnalysisException(s"Can't add $attr to ${a.simpleString}")
- }
- }
- val newAggregateExpressions = a.aggregateExpressions ++ missingAttrs
- a.copy(aggregateExpressions = newAggregateExpressions)
- case g: Generate =>
- // If join is false, we will convert it to true for getting from the child the missing
- // attributes that its child might have or could have.
- val missing = missingAttrs -- g.child.outputSet
- g.copy(join = true, child = addMissingAttr(g.child, missing))
- case d: Distinct =>
- throw new AnalysisException(s"Can't add $missingAttrs to $d")
- case u: UnaryNode =>
- u.withNewChildren(addMissingAttr(u.child, missingAttrs) :: Nil)
- case other =>
- throw new AnalysisException(s"Can't add $missingAttrs to $other")
- }
- }
-
- /**
- * Resolve the expression on a specified logical plan and it's child (recursively), until
- * the expression is resolved or meet a non-unary node or Subquery.
- */
- @tailrec
- private def resolveExpressionRecursively(expr: Expression, plan: LogicalPlan): Expression = {
- val resolved = resolveExpression(expr, plan)
- if (resolved.resolved) {
- resolved
+ private def resolveExprsAndAddMissingAttrs(
--- End diff --
I refactored the code to resolve expressions and add missing attributes in one shot, so that we have a central place to deal with analysis barrier and to decide which operator is supported and which is not.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark issue #20094: [SPARK-20392][SQL][followup] should not add extra Analys...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the issue:
https://github.com/apache/spark/pull/20094
**[Test build #85450 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/85450/testReport)** for PR 20094 at commit [`cd39760`](https://github.com/apache/spark/commit/cd397605eaf81e3725396c715602734596a83ac3).
* This patch **fails PySpark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org