You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by davies <gi...@git.apache.org> on 2016/02/10 19:21:46 UTC
[GitHub] spark pull request: [SPARK-12705] [SQL] push missing attributes in...
GitHub user davies opened a pull request:
https://github.com/apache/spark/pull/11153
[SPARK-12705] [SQL] push missing attributes in Sort
The current implementation of ResolveSortReferences can only push one missing attributes into it's child, it failed to analyze TPCDS Q98, because of there are two missing attributes in that (one from Window, another from Aggregate).
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/davies/spark resolve_sort
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/spark/pull/11153.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 #11153
----
commit e1919ccf1e0a878bfb320d41beeffbc7f4c9b5da
Author: Davies Liu <da...@databricks.com>
Date: 2016-02-10T18:18:11Z
push missing attributes in Sort
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12705] [SQL] push missing attributes fo...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on a diff in the pull request:
https://github.com/apache/spark/pull/11153#discussion_r52707329
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala ---
@@ -572,98 +572,64 @@ class Analyzer(
// Skip sort with aggregate. This will be handled in ResolveAggregateFunctions
case sa @ Sort(_, _, child: Aggregate) => sa
- case s @ Sort(_, _, child) if !s.resolved && child.resolved =>
- val (newOrdering, missingResolvableAttrs) = collectResolvableMissingAttrs(s.order, child)
-
- if (missingResolvableAttrs.isEmpty) {
- val unresolvableAttrs = s.order.filterNot(_.resolved)
- logDebug(s"Failed to find $unresolvableAttrs in ${child.output.mkString(", ")}")
- s // Nothing we can do here. Return original plan.
- } else {
- // Add the missing attributes into projectList of Project/Window or
- // aggregateExpressions of Aggregate, if they are in the inputSet
- // but not in the outputSet of the plan.
- val newChild = child transformUp {
- case p: Project =>
- p.copy(projectList = p.projectList ++
- missingResolvableAttrs.filter((p.inputSet -- p.outputSet).contains))
- case w: Window =>
- w.copy(projectList = w.projectList ++
- missingResolvableAttrs.filter((w.inputSet -- w.outputSet).contains))
- case a: Aggregate =>
- val resolvableAttrs = missingResolvableAttrs.filter(a.groupingExpressions.contains)
- val notResolvedAttrs = resolvableAttrs.filterNot(a.aggregateExpressions.contains)
- val newAggregateExpressions = a.aggregateExpressions ++ notResolvedAttrs
- a.copy(aggregateExpressions = newAggregateExpressions)
- case o => o
- }
-
+ case s @ Sort(order, _, child) if !s.resolved && child.resolved =>
+ 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(newOrdering, s.global, newChild))
+ Sort(newOrder, s.global, addMissingAttr(child, missingAttrs)))
+ } else if (newOrder != order) {
+ s.copy(order = newOrder)
+ } else {
+ s
}
}
/**
- * Traverse the tree until resolving the sorting attributes
- * Return all the resolvable missing sorting attributes
- */
- @tailrec
- private def collectResolvableMissingAttrs(
- ordering: Seq[SortOrder],
- plan: LogicalPlan): (Seq[SortOrder], Seq[Attribute]) = {
+ * Add the missing attributes into projectList of Project/Window or aggregateExpressions of
+ * Aggregate.
+ */
+ private def addMissingAttr(plan: LogicalPlan, missingAttrs: AttributeSet): LogicalPlan = {
--- End diff --
It makes sense to me. Thanks!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12705] [SQL] push missing attributes in...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11153#issuecomment-182556937
Merged build finished. Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12705] [SQL] push missing attributes fo...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on a diff in the pull request:
https://github.com/apache/spark/pull/11153#discussion_r52628130
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala ---
@@ -572,98 +572,64 @@ class Analyzer(
// Skip sort with aggregate. This will be handled in ResolveAggregateFunctions
case sa @ Sort(_, _, child: Aggregate) => sa
- case s @ Sort(_, _, child) if !s.resolved && child.resolved =>
- val (newOrdering, missingResolvableAttrs) = collectResolvableMissingAttrs(s.order, child)
-
- if (missingResolvableAttrs.isEmpty) {
- val unresolvableAttrs = s.order.filterNot(_.resolved)
- logDebug(s"Failed to find $unresolvableAttrs in ${child.output.mkString(", ")}")
- s // Nothing we can do here. Return original plan.
- } else {
- // Add the missing attributes into projectList of Project/Window or
- // aggregateExpressions of Aggregate, if they are in the inputSet
- // but not in the outputSet of the plan.
- val newChild = child transformUp {
- case p: Project =>
- p.copy(projectList = p.projectList ++
- missingResolvableAttrs.filter((p.inputSet -- p.outputSet).contains))
- case w: Window =>
- w.copy(projectList = w.projectList ++
- missingResolvableAttrs.filter((w.inputSet -- w.outputSet).contains))
- case a: Aggregate =>
- val resolvableAttrs = missingResolvableAttrs.filter(a.groupingExpressions.contains)
- val notResolvedAttrs = resolvableAttrs.filterNot(a.aggregateExpressions.contains)
- val newAggregateExpressions = a.aggregateExpressions ++ notResolvedAttrs
- a.copy(aggregateExpressions = newAggregateExpressions)
- case o => o
- }
-
+ case s @ Sort(order, _, child) if !s.resolved && child.resolved =>
+ 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(newOrdering, s.global, newChild))
+ Sort(newOrder, s.global, addMissingAttr(child, missingAttrs)))
+ } else if (newOrder != order) {
+ s.copy(order = newOrder)
+ } else {
+ s
}
}
/**
- * Traverse the tree until resolving the sorting attributes
- * Return all the resolvable missing sorting attributes
- */
- @tailrec
- private def collectResolvableMissingAttrs(
- ordering: Seq[SortOrder],
- plan: LogicalPlan): (Seq[SortOrder], Seq[Attribute]) = {
+ * Add the missing attributes into projectList of Project/Window or aggregateExpressions of
+ * Aggregate.
+ */
+ private def addMissingAttr(plan: LogicalPlan, missingAttrs: AttributeSet): LogicalPlan = {
--- End diff --
This is not using tail recursion. When the tree is large, we might hit stack overflow. I am fine if this is not a concern anymore.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12705] [SQL] push missing attributes fo...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on a diff in the pull request:
https://github.com/apache/spark/pull/11153#discussion_r52707357
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala ---
@@ -572,98 +572,64 @@ class Analyzer(
// Skip sort with aggregate. This will be handled in ResolveAggregateFunctions
case sa @ Sort(_, _, child: Aggregate) => sa
- case s @ Sort(_, _, child) if !s.resolved && child.resolved =>
- val (newOrdering, missingResolvableAttrs) = collectResolvableMissingAttrs(s.order, child)
-
- if (missingResolvableAttrs.isEmpty) {
- val unresolvableAttrs = s.order.filterNot(_.resolved)
- logDebug(s"Failed to find $unresolvableAttrs in ${child.output.mkString(", ")}")
- s // Nothing we can do here. Return original plan.
- } else {
- // Add the missing attributes into projectList of Project/Window or
- // aggregateExpressions of Aggregate, if they are in the inputSet
- // but not in the outputSet of the plan.
- val newChild = child transformUp {
- case p: Project =>
- p.copy(projectList = p.projectList ++
- missingResolvableAttrs.filter((p.inputSet -- p.outputSet).contains))
- case w: Window =>
- w.copy(projectList = w.projectList ++
- missingResolvableAttrs.filter((w.inputSet -- w.outputSet).contains))
- case a: Aggregate =>
- val resolvableAttrs = missingResolvableAttrs.filter(a.groupingExpressions.contains)
- val notResolvedAttrs = resolvableAttrs.filterNot(a.aggregateExpressions.contains)
- val newAggregateExpressions = a.aggregateExpressions ++ notResolvedAttrs
- a.copy(aggregateExpressions = newAggregateExpressions)
- case o => o
- }
-
+ case s @ Sort(order, _, child) if !s.resolved && child.resolved =>
+ 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(newOrdering, s.global, newChild))
+ Sort(newOrder, s.global, addMissingAttr(child, missingAttrs)))
+ } else if (newOrder != order) {
+ s.copy(order = newOrder)
+ } else {
+ s
}
}
/**
- * Traverse the tree until resolving the sorting attributes
- * Return all the resolvable missing sorting attributes
- */
- @tailrec
- private def collectResolvableMissingAttrs(
- ordering: Seq[SortOrder],
- plan: LogicalPlan): (Seq[SortOrder], Seq[Attribute]) = {
+ * 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 plan
+ }
plan match {
- // Only Windows and Project have projectList-like attribute.
- case un: UnaryNode if un.isInstanceOf[Project] || un.isInstanceOf[Window] =>
- val (newOrdering, missingAttrs) = resolveAndFindMissing(ordering, un, un.child)
- // If missingAttrs is non empty, that means we got it and return it;
- // Otherwise, continue to traverse the tree.
- if (missingAttrs.nonEmpty) {
- (newOrdering, missingAttrs)
- } else {
- collectResolvableMissingAttrs(ordering, un.child)
- }
+ case p: Project =>
+ val missing = missingAttrs -- p.child.outputSet
+ Project(p.projectList ++ missingAttrs, addMissingAttr(p.child, missing))
+ case w: Window =>
+ val missing = missingAttrs -- w.child.outputSet
+ w.copy(projectList = w.projectList ++ missingAttrs,
+ child = addMissingAttr(w.child, missing))
case a: Aggregate =>
- val (newOrdering, missingAttrs) = resolveAndFindMissing(ordering, a, a.child)
- // For Aggregate, all the order by columns must be specified in group by clauses
- if (missingAttrs.nonEmpty &&
- missingAttrs.forall(ar => a.groupingExpressions.exists(_.semanticEquals(ar)))) {
- (newOrdering, missingAttrs)
- } else {
- // If missingAttrs is empty, we are unable to resolve any unresolved missing attributes
- (Seq.empty[SortOrder], Seq.empty[Attribute])
+ // all the missing attributes should be grouping expressions
+ // TODO: push down AggregateExpression
+ missingAttrs.foreach { attr =>
+ if (!a.groupingExpressions.contains(attr)) {
+ throw new AnalysisException(s"Can't add $attr to ${a.simpleString}")
+ }
}
- // Jump over the following UnaryNode types
- // The output of these types is the same as their child's output
- case _: Distinct |
- _: Filter |
- _: RepartitionByExpression =>
- collectResolvableMissingAttrs(ordering, plan.asInstanceOf[UnaryNode].child)
- // If hitting the other unsupported operators, we are unable to resolve it.
- case other => (Seq.empty[SortOrder], Seq.empty[Attribute])
+ val newAggregateExpressions = a.aggregateExpressions ++ missingAttrs
+ a.copy(aggregateExpressions = newAggregateExpressions)
+ case u: UnaryNode =>
--- End diff --
Sure, will do.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12705] [SQL] push missing attributes fo...
Posted by cloud-fan <gi...@git.apache.org>.
Github user cloud-fan commented on the pull request:
https://github.com/apache/spark/pull/11153#issuecomment-183238642
LGTM except one comment
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12705] [SQL] push missing attributes fo...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11153#issuecomment-182752763
**[Test build #51085 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/51085/consoleFull)** for PR 11153 at commit [`c4607dd`](https://github.com/apache/spark/commit/c4607dd6f08801ff117768a89e3d13bb748b2d43).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12705] [SQL] push missing attributes fo...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on the pull request:
https://github.com/apache/spark/pull/11153#issuecomment-182720078
cc @gatorsmile
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12705] [SQL] push missing attributes in...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11153#issuecomment-182556217
**[Test build #51040 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/51040/consoleFull)** for PR 11153 at commit [`e1919cc`](https://github.com/apache/spark/commit/e1919ccf1e0a878bfb320d41beeffbc7f4c9b5da).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12705] [SQL] push missing attributes in...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11153#issuecomment-182556945
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/51040/
Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12705] [SQL] push missing attributes fo...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on a diff in the pull request:
https://github.com/apache/spark/pull/11153#discussion_r52628550
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala ---
@@ -572,98 +572,64 @@ class Analyzer(
// Skip sort with aggregate. This will be handled in ResolveAggregateFunctions
case sa @ Sort(_, _, child: Aggregate) => sa
- case s @ Sort(_, _, child) if !s.resolved && child.resolved =>
- val (newOrdering, missingResolvableAttrs) = collectResolvableMissingAttrs(s.order, child)
-
- if (missingResolvableAttrs.isEmpty) {
- val unresolvableAttrs = s.order.filterNot(_.resolved)
- logDebug(s"Failed to find $unresolvableAttrs in ${child.output.mkString(", ")}")
- s // Nothing we can do here. Return original plan.
- } else {
- // Add the missing attributes into projectList of Project/Window or
- // aggregateExpressions of Aggregate, if they are in the inputSet
- // but not in the outputSet of the plan.
- val newChild = child transformUp {
- case p: Project =>
- p.copy(projectList = p.projectList ++
- missingResolvableAttrs.filter((p.inputSet -- p.outputSet).contains))
- case w: Window =>
- w.copy(projectList = w.projectList ++
- missingResolvableAttrs.filter((w.inputSet -- w.outputSet).contains))
- case a: Aggregate =>
- val resolvableAttrs = missingResolvableAttrs.filter(a.groupingExpressions.contains)
- val notResolvedAttrs = resolvableAttrs.filterNot(a.aggregateExpressions.contains)
- val newAggregateExpressions = a.aggregateExpressions ++ notResolvedAttrs
- a.copy(aggregateExpressions = newAggregateExpressions)
- case o => o
- }
-
+ case s @ Sort(order, _, child) if !s.resolved && child.resolved =>
+ 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(newOrdering, s.global, newChild))
+ Sort(newOrder, s.global, addMissingAttr(child, missingAttrs)))
+ } else if (newOrder != order) {
+ s.copy(order = newOrder)
+ } else {
+ s
}
}
/**
- * Traverse the tree until resolving the sorting attributes
- * Return all the resolvable missing sorting attributes
- */
- @tailrec
- private def collectResolvableMissingAttrs(
- ordering: Seq[SortOrder],
- plan: LogicalPlan): (Seq[SortOrder], Seq[Attribute]) = {
+ * 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 plan
+ }
plan match {
- // Only Windows and Project have projectList-like attribute.
- case un: UnaryNode if un.isInstanceOf[Project] || un.isInstanceOf[Window] =>
- val (newOrdering, missingAttrs) = resolveAndFindMissing(ordering, un, un.child)
- // If missingAttrs is non empty, that means we got it and return it;
- // Otherwise, continue to traverse the tree.
- if (missingAttrs.nonEmpty) {
- (newOrdering, missingAttrs)
- } else {
- collectResolvableMissingAttrs(ordering, un.child)
- }
+ case p: Project =>
+ val missing = missingAttrs -- p.child.outputSet
+ Project(p.projectList ++ missingAttrs, addMissingAttr(p.child, missing))
+ case w: Window =>
+ val missing = missingAttrs -- w.child.outputSet
+ w.copy(projectList = w.projectList ++ missingAttrs,
+ child = addMissingAttr(w.child, missing))
case a: Aggregate =>
- val (newOrdering, missingAttrs) = resolveAndFindMissing(ordering, a, a.child)
- // For Aggregate, all the order by columns must be specified in group by clauses
- if (missingAttrs.nonEmpty &&
- missingAttrs.forall(ar => a.groupingExpressions.exists(_.semanticEquals(ar)))) {
- (newOrdering, missingAttrs)
- } else {
- // If missingAttrs is empty, we are unable to resolve any unresolved missing attributes
- (Seq.empty[SortOrder], Seq.empty[Attribute])
+ // all the missing attributes should be grouping expressions
+ // TODO: push down AggregateExpression
+ missingAttrs.foreach { attr =>
+ if (!a.groupingExpressions.contains(attr)) {
+ throw new AnalysisException(s"Can't add $attr to ${a.simpleString}")
+ }
}
- // Jump over the following UnaryNode types
- // The output of these types is the same as their child's output
- case _: Distinct |
- _: Filter |
- _: RepartitionByExpression =>
- collectResolvableMissingAttrs(ordering, plan.asInstanceOf[UnaryNode].child)
- // If hitting the other unsupported operators, we are unable to resolve it.
- case other => (Seq.empty[SortOrder], Seq.empty[Attribute])
+ val newAggregateExpressions = a.aggregateExpressions ++ missingAttrs
+ a.copy(aggregateExpressions = newAggregateExpressions)
+ case u: UnaryNode =>
--- End diff --
If needed, I can add more test cases after this is merged, since it could be time-consuming. Please feel free to let me know.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12705] [SQL] push missing attributes fo...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/11153#discussion_r52648798
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala ---
@@ -572,98 +572,64 @@ class Analyzer(
// Skip sort with aggregate. This will be handled in ResolveAggregateFunctions
case sa @ Sort(_, _, child: Aggregate) => sa
- case s @ Sort(_, _, child) if !s.resolved && child.resolved =>
- val (newOrdering, missingResolvableAttrs) = collectResolvableMissingAttrs(s.order, child)
-
- if (missingResolvableAttrs.isEmpty) {
- val unresolvableAttrs = s.order.filterNot(_.resolved)
- logDebug(s"Failed to find $unresolvableAttrs in ${child.output.mkString(", ")}")
- s // Nothing we can do here. Return original plan.
- } else {
- // Add the missing attributes into projectList of Project/Window or
- // aggregateExpressions of Aggregate, if they are in the inputSet
- // but not in the outputSet of the plan.
- val newChild = child transformUp {
- case p: Project =>
- p.copy(projectList = p.projectList ++
- missingResolvableAttrs.filter((p.inputSet -- p.outputSet).contains))
- case w: Window =>
- w.copy(projectList = w.projectList ++
- missingResolvableAttrs.filter((w.inputSet -- w.outputSet).contains))
- case a: Aggregate =>
- val resolvableAttrs = missingResolvableAttrs.filter(a.groupingExpressions.contains)
- val notResolvedAttrs = resolvableAttrs.filterNot(a.aggregateExpressions.contains)
- val newAggregateExpressions = a.aggregateExpressions ++ notResolvedAttrs
- a.copy(aggregateExpressions = newAggregateExpressions)
- case o => o
- }
-
+ case s @ Sort(order, _, child) if !s.resolved && child.resolved =>
+ 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(newOrdering, s.global, newChild))
+ Sort(newOrder, s.global, addMissingAttr(child, missingAttrs)))
+ } else if (newOrder != order) {
+ s.copy(order = newOrder)
+ } else {
+ s
}
}
/**
- * Traverse the tree until resolving the sorting attributes
- * Return all the resolvable missing sorting attributes
- */
- @tailrec
- private def collectResolvableMissingAttrs(
- ordering: Seq[SortOrder],
- plan: LogicalPlan): (Seq[SortOrder], Seq[Attribute]) = {
+ * 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 plan
+ }
plan match {
- // Only Windows and Project have projectList-like attribute.
- case un: UnaryNode if un.isInstanceOf[Project] || un.isInstanceOf[Window] =>
- val (newOrdering, missingAttrs) = resolveAndFindMissing(ordering, un, un.child)
- // If missingAttrs is non empty, that means we got it and return it;
- // Otherwise, continue to traverse the tree.
- if (missingAttrs.nonEmpty) {
- (newOrdering, missingAttrs)
- } else {
- collectResolvableMissingAttrs(ordering, un.child)
- }
+ case p: Project =>
+ val missing = missingAttrs -- p.child.outputSet
+ Project(p.projectList ++ missingAttrs, addMissingAttr(p.child, missing))
+ case w: Window =>
+ val missing = missingAttrs -- w.child.outputSet
+ w.copy(projectList = w.projectList ++ missingAttrs,
+ child = addMissingAttr(w.child, missing))
case a: Aggregate =>
- val (newOrdering, missingAttrs) = resolveAndFindMissing(ordering, a, a.child)
- // For Aggregate, all the order by columns must be specified in group by clauses
- if (missingAttrs.nonEmpty &&
- missingAttrs.forall(ar => a.groupingExpressions.exists(_.semanticEquals(ar)))) {
- (newOrdering, missingAttrs)
- } else {
- // If missingAttrs is empty, we are unable to resolve any unresolved missing attributes
- (Seq.empty[SortOrder], Seq.empty[Attribute])
+ // all the missing attributes should be grouping expressions
+ // TODO: push down AggregateExpression
+ missingAttrs.foreach { attr =>
+ if (!a.groupingExpressions.contains(attr)) {
+ throw new AnalysisException(s"Can't add $attr to ${a.simpleString}")
+ }
}
- // Jump over the following UnaryNode types
- // The output of these types is the same as their child's output
- case _: Distinct |
- _: Filter |
- _: RepartitionByExpression =>
- collectResolvableMissingAttrs(ordering, plan.asInstanceOf[UnaryNode].child)
- // If hitting the other unsupported operators, we are unable to resolve it.
- case other => (Seq.empty[SortOrder], Seq.empty[Attribute])
+ val newAggregateExpressions = a.aggregateExpressions ++ missingAttrs
+ a.copy(aggregateExpressions = newAggregateExpressions)
+ case u: UnaryNode =>
--- End diff --
Will exclude Subquery, feel free to add more tests.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12705] [SQL] push missing attributes fo...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on the pull request:
https://github.com/apache/spark/pull/11153#issuecomment-183422964
Merging this into master, thanks!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12705] [SQL] push missing attributes fo...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on a diff in the pull request:
https://github.com/apache/spark/pull/11153#discussion_r52627258
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala ---
@@ -572,98 +572,64 @@ class Analyzer(
// Skip sort with aggregate. This will be handled in ResolveAggregateFunctions
case sa @ Sort(_, _, child: Aggregate) => sa
- case s @ Sort(_, _, child) if !s.resolved && child.resolved =>
- val (newOrdering, missingResolvableAttrs) = collectResolvableMissingAttrs(s.order, child)
-
- if (missingResolvableAttrs.isEmpty) {
- val unresolvableAttrs = s.order.filterNot(_.resolved)
- logDebug(s"Failed to find $unresolvableAttrs in ${child.output.mkString(", ")}")
- s // Nothing we can do here. Return original plan.
- } else {
- // Add the missing attributes into projectList of Project/Window or
- // aggregateExpressions of Aggregate, if they are in the inputSet
- // but not in the outputSet of the plan.
- val newChild = child transformUp {
- case p: Project =>
- p.copy(projectList = p.projectList ++
- missingResolvableAttrs.filter((p.inputSet -- p.outputSet).contains))
- case w: Window =>
- w.copy(projectList = w.projectList ++
- missingResolvableAttrs.filter((w.inputSet -- w.outputSet).contains))
- case a: Aggregate =>
- val resolvableAttrs = missingResolvableAttrs.filter(a.groupingExpressions.contains)
- val notResolvedAttrs = resolvableAttrs.filterNot(a.aggregateExpressions.contains)
- val newAggregateExpressions = a.aggregateExpressions ++ notResolvedAttrs
- a.copy(aggregateExpressions = newAggregateExpressions)
- case o => o
- }
-
+ case s @ Sort(order, _, child) if !s.resolved && child.resolved =>
+ val newOrder = order.map(resolveExpressionRecursively(_, child).asInstanceOf[SortOrder])
+ val requiredAttrs = AttributeSet(newOrder).filter(_.resolved)
+ val missingAttrs = requiredAttrs -- child.outputSet
--- End diff --
Since this solution could skip `Subquery`, we might need to add/change `qualifiers` for the missingAttrs, if necessary.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12705] [SQL] push missing attributes fo...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on a diff in the pull request:
https://github.com/apache/spark/pull/11153#discussion_r52627598
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala ---
@@ -572,98 +572,64 @@ class Analyzer(
// Skip sort with aggregate. This will be handled in ResolveAggregateFunctions
case sa @ Sort(_, _, child: Aggregate) => sa
- case s @ Sort(_, _, child) if !s.resolved && child.resolved =>
- val (newOrdering, missingResolvableAttrs) = collectResolvableMissingAttrs(s.order, child)
-
- if (missingResolvableAttrs.isEmpty) {
- val unresolvableAttrs = s.order.filterNot(_.resolved)
- logDebug(s"Failed to find $unresolvableAttrs in ${child.output.mkString(", ")}")
- s // Nothing we can do here. Return original plan.
- } else {
- // Add the missing attributes into projectList of Project/Window or
- // aggregateExpressions of Aggregate, if they are in the inputSet
- // but not in the outputSet of the plan.
- val newChild = child transformUp {
- case p: Project =>
- p.copy(projectList = p.projectList ++
- missingResolvableAttrs.filter((p.inputSet -- p.outputSet).contains))
- case w: Window =>
- w.copy(projectList = w.projectList ++
- missingResolvableAttrs.filter((w.inputSet -- w.outputSet).contains))
- case a: Aggregate =>
- val resolvableAttrs = missingResolvableAttrs.filter(a.groupingExpressions.contains)
- val notResolvedAttrs = resolvableAttrs.filterNot(a.aggregateExpressions.contains)
- val newAggregateExpressions = a.aggregateExpressions ++ notResolvedAttrs
- a.copy(aggregateExpressions = newAggregateExpressions)
- case o => o
- }
-
+ case s @ Sort(order, _, child) if !s.resolved && child.resolved =>
+ 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(newOrdering, s.global, newChild))
+ Sort(newOrder, s.global, addMissingAttr(child, missingAttrs)))
+ } else if (newOrder != order) {
+ s.copy(order = newOrder)
+ } else {
+ s
}
}
/**
- * Traverse the tree until resolving the sorting attributes
- * Return all the resolvable missing sorting attributes
- */
- @tailrec
- private def collectResolvableMissingAttrs(
- ordering: Seq[SortOrder],
- plan: LogicalPlan): (Seq[SortOrder], Seq[Attribute]) = {
+ * 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 plan
+ }
plan match {
- // Only Windows and Project have projectList-like attribute.
- case un: UnaryNode if un.isInstanceOf[Project] || un.isInstanceOf[Window] =>
- val (newOrdering, missingAttrs) = resolveAndFindMissing(ordering, un, un.child)
- // If missingAttrs is non empty, that means we got it and return it;
- // Otherwise, continue to traverse the tree.
- if (missingAttrs.nonEmpty) {
- (newOrdering, missingAttrs)
- } else {
- collectResolvableMissingAttrs(ordering, un.child)
- }
+ case p: Project =>
+ val missing = missingAttrs -- p.child.outputSet
+ Project(p.projectList ++ missingAttrs, addMissingAttr(p.child, missing))
+ case w: Window =>
+ val missing = missingAttrs -- w.child.outputSet
+ w.copy(projectList = w.projectList ++ missingAttrs,
+ child = addMissingAttr(w.child, missing))
case a: Aggregate =>
- val (newOrdering, missingAttrs) = resolveAndFindMissing(ordering, a, a.child)
- // For Aggregate, all the order by columns must be specified in group by clauses
- if (missingAttrs.nonEmpty &&
- missingAttrs.forall(ar => a.groupingExpressions.exists(_.semanticEquals(ar)))) {
- (newOrdering, missingAttrs)
- } else {
- // If missingAttrs is empty, we are unable to resolve any unresolved missing attributes
- (Seq.empty[SortOrder], Seq.empty[Attribute])
+ // all the missing attributes should be grouping expressions
+ // TODO: push down AggregateExpression
+ missingAttrs.foreach { attr =>
+ if (!a.groupingExpressions.contains(attr)) {
--- End diff --
`contains` will call `equals` instead of `semanticEquals`. Thus, it might not work if we cross `Subquery`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12705] [SQL] push missing attributes fo...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11153#issuecomment-183057571
**[Test build #51124 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/51124/consoleFull)** for PR 11153 at commit [`dce3857`](https://github.com/apache/spark/commit/dce38575730ae92972d8c15e4a6d13983eeb0392).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12705] [SQL] push missing attributes fo...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11153#issuecomment-183058010
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/51124/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12705] [SQL] push missing attributes fo...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11153#issuecomment-183058007
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12705] [SQL] push missing attributes fo...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11153#issuecomment-182753089
Merged build finished. Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12705] [SQL] push missing attributes fo...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on a diff in the pull request:
https://github.com/apache/spark/pull/11153#discussion_r52626550
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala ---
@@ -572,98 +572,64 @@ class Analyzer(
// Skip sort with aggregate. This will be handled in ResolveAggregateFunctions
case sa @ Sort(_, _, child: Aggregate) => sa
- case s @ Sort(_, _, child) if !s.resolved && child.resolved =>
- val (newOrdering, missingResolvableAttrs) = collectResolvableMissingAttrs(s.order, child)
-
- if (missingResolvableAttrs.isEmpty) {
- val unresolvableAttrs = s.order.filterNot(_.resolved)
- logDebug(s"Failed to find $unresolvableAttrs in ${child.output.mkString(", ")}")
- s // Nothing we can do here. Return original plan.
- } else {
- // Add the missing attributes into projectList of Project/Window or
- // aggregateExpressions of Aggregate, if they are in the inputSet
- // but not in the outputSet of the plan.
- val newChild = child transformUp {
- case p: Project =>
- p.copy(projectList = p.projectList ++
- missingResolvableAttrs.filter((p.inputSet -- p.outputSet).contains))
- case w: Window =>
- w.copy(projectList = w.projectList ++
- missingResolvableAttrs.filter((w.inputSet -- w.outputSet).contains))
- case a: Aggregate =>
- val resolvableAttrs = missingResolvableAttrs.filter(a.groupingExpressions.contains)
- val notResolvedAttrs = resolvableAttrs.filterNot(a.aggregateExpressions.contains)
- val newAggregateExpressions = a.aggregateExpressions ++ notResolvedAttrs
- a.copy(aggregateExpressions = newAggregateExpressions)
- case o => o
- }
-
+ case s @ Sort(order, _, child) if !s.resolved && child.resolved =>
+ 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(newOrdering, s.global, newChild))
+ Sort(newOrder, s.global, addMissingAttr(child, missingAttrs)))
+ } else if (newOrder != order) {
+ s.copy(order = newOrder)
+ } else {
+ s
}
}
/**
- * Traverse the tree until resolving the sorting attributes
- * Return all the resolvable missing sorting attributes
- */
- @tailrec
- private def collectResolvableMissingAttrs(
- ordering: Seq[SortOrder],
- plan: LogicalPlan): (Seq[SortOrder], Seq[Attribute]) = {
+ * 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 plan
+ }
plan match {
- // Only Windows and Project have projectList-like attribute.
- case un: UnaryNode if un.isInstanceOf[Project] || un.isInstanceOf[Window] =>
- val (newOrdering, missingAttrs) = resolveAndFindMissing(ordering, un, un.child)
- // If missingAttrs is non empty, that means we got it and return it;
- // Otherwise, continue to traverse the tree.
- if (missingAttrs.nonEmpty) {
- (newOrdering, missingAttrs)
- } else {
- collectResolvableMissingAttrs(ordering, un.child)
- }
+ case p: Project =>
+ val missing = missingAttrs -- p.child.outputSet
+ Project(p.projectList ++ missingAttrs, addMissingAttr(p.child, missing))
+ case w: Window =>
+ val missing = missingAttrs -- w.child.outputSet
+ w.copy(projectList = w.projectList ++ missingAttrs,
+ child = addMissingAttr(w.child, missing))
case a: Aggregate =>
- val (newOrdering, missingAttrs) = resolveAndFindMissing(ordering, a, a.child)
- // For Aggregate, all the order by columns must be specified in group by clauses
- if (missingAttrs.nonEmpty &&
- missingAttrs.forall(ar => a.groupingExpressions.exists(_.semanticEquals(ar)))) {
- (newOrdering, missingAttrs)
- } else {
- // If missingAttrs is empty, we are unable to resolve any unresolved missing attributes
- (Seq.empty[SortOrder], Seq.empty[Attribute])
+ // all the missing attributes should be grouping expressions
+ // TODO: push down AggregateExpression
+ missingAttrs.foreach { attr =>
+ if (!a.groupingExpressions.contains(attr)) {
+ throw new AnalysisException(s"Can't add $attr to ${a.simpleString}")
+ }
}
- // Jump over the following UnaryNode types
- // The output of these types is the same as their child's output
- case _: Distinct |
- _: Filter |
- _: RepartitionByExpression =>
- collectResolvableMissingAttrs(ordering, plan.asInstanceOf[UnaryNode].child)
- // If hitting the other unsupported operators, we are unable to resolve it.
- case other => (Seq.empty[SortOrder], Seq.empty[Attribute])
+ val newAggregateExpressions = a.aggregateExpressions ++ missingAttrs
+ a.copy(aggregateExpressions = newAggregateExpressions)
+ case u: UnaryNode =>
--- End diff --
It sounds like we want to cover all the `UnaryNode` here? This is different from what we discussed. I am fine if you want to add more supports (e.g., crossing the boundary of subquery), but we might need to add more test cases to ensure it does not break anything.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12705] [SQL] push missing attributes fo...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11153#issuecomment-182728121
**[Test build #51085 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/51085/consoleFull)** for PR 11153 at commit [`c4607dd`](https://github.com/apache/spark/commit/c4607dd6f08801ff117768a89e3d13bb748b2d43).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12705] [SQL] push missing attributes in...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11153#issuecomment-182581239
Merged build finished. Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12705] [SQL] push missing attributes fo...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/spark/pull/11153
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12705] [SQL] push missing attributes in...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on the pull request:
https://github.com/apache/spark/pull/11153#issuecomment-182514659
cc @marmbrus
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12705] [SQL] push missing attributes fo...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/11153#discussion_r52646199
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala ---
@@ -572,98 +572,64 @@ class Analyzer(
// Skip sort with aggregate. This will be handled in ResolveAggregateFunctions
case sa @ Sort(_, _, child: Aggregate) => sa
- case s @ Sort(_, _, child) if !s.resolved && child.resolved =>
- val (newOrdering, missingResolvableAttrs) = collectResolvableMissingAttrs(s.order, child)
-
- if (missingResolvableAttrs.isEmpty) {
- val unresolvableAttrs = s.order.filterNot(_.resolved)
- logDebug(s"Failed to find $unresolvableAttrs in ${child.output.mkString(", ")}")
- s // Nothing we can do here. Return original plan.
- } else {
- // Add the missing attributes into projectList of Project/Window or
- // aggregateExpressions of Aggregate, if they are in the inputSet
- // but not in the outputSet of the plan.
- val newChild = child transformUp {
- case p: Project =>
- p.copy(projectList = p.projectList ++
- missingResolvableAttrs.filter((p.inputSet -- p.outputSet).contains))
- case w: Window =>
- w.copy(projectList = w.projectList ++
- missingResolvableAttrs.filter((w.inputSet -- w.outputSet).contains))
- case a: Aggregate =>
- val resolvableAttrs = missingResolvableAttrs.filter(a.groupingExpressions.contains)
- val notResolvedAttrs = resolvableAttrs.filterNot(a.aggregateExpressions.contains)
- val newAggregateExpressions = a.aggregateExpressions ++ notResolvedAttrs
- a.copy(aggregateExpressions = newAggregateExpressions)
- case o => o
- }
-
+ case s @ Sort(order, _, child) if !s.resolved && child.resolved =>
+ 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(newOrdering, s.global, newChild))
+ Sort(newOrder, s.global, addMissingAttr(child, missingAttrs)))
+ } else if (newOrder != order) {
+ s.copy(order = newOrder)
+ } else {
+ s
}
}
/**
- * Traverse the tree until resolving the sorting attributes
- * Return all the resolvable missing sorting attributes
- */
- @tailrec
- private def collectResolvableMissingAttrs(
- ordering: Seq[SortOrder],
- plan: LogicalPlan): (Seq[SortOrder], Seq[Attribute]) = {
+ * Add the missing attributes into projectList of Project/Window or aggregateExpressions of
+ * Aggregate.
+ */
+ private def addMissingAttr(plan: LogicalPlan, missingAttrs: AttributeSet): LogicalPlan = {
--- End diff --
Because it can not go over JOIN, it's very uncommon to have thousands of unary nodes in practice.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12705] [SQL] push missing attributes fo...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on the pull request:
https://github.com/apache/spark/pull/11153#issuecomment-182736681
Sorry, my previous fix does not cover all the scenarios. Will read your fix tomorrow. Thanks!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12705] [SQL] push missing attributes fo...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on a diff in the pull request:
https://github.com/apache/spark/pull/11153#discussion_r52629031
--- Diff: sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisSuite.scala ---
@@ -90,7 +90,7 @@ class AnalysisSuite extends AnalysisTest {
.where(a > "str").select(a, b, c)
.where(b > "str").select(a, b, c)
.sortBy(b.asc, c.desc)
- .select(a, b).select(a)
--- End diff --
Based on this test case, it sounds like the previous PR can cover the case of two missing attributes. Do you know why Q98 still has an issue? Thanks!
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12705] [SQL] push missing attributes fo...
Posted by davies <gi...@git.apache.org>.
Github user davies commented on a diff in the pull request:
https://github.com/apache/spark/pull/11153#discussion_r52645515
--- Diff: sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/AnalysisSuite.scala ---
@@ -90,7 +90,7 @@ class AnalysisSuite extends AnalysisTest {
.where(a > "str").select(a, b, c)
.where(b > "str").select(a, b, c)
.sortBy(b.asc, c.desc)
- .select(a, b).select(a)
--- End diff --
It depends on what the missing attributes are, checkout the added regression test.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12705] [SQL] push missing attributes fo...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11153#issuecomment-183013099
**[Test build #51124 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/51124/consoleFull)** for PR 11153 at commit [`dce3857`](https://github.com/apache/spark/commit/dce38575730ae92972d8c15e4a6d13983eeb0392).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12705] [SQL] push missing attributes in...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11153#issuecomment-182581242
Test FAILed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/51050/
Test FAILed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12705] [SQL] push missing attributes fo...
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/11153#discussion_r52717321
--- Diff: sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala ---
@@ -572,98 +572,64 @@ class Analyzer(
// Skip sort with aggregate. This will be handled in ResolveAggregateFunctions
case sa @ Sort(_, _, child: Aggregate) => sa
- case s @ Sort(_, _, child) if !s.resolved && child.resolved =>
- val (newOrdering, missingResolvableAttrs) = collectResolvableMissingAttrs(s.order, child)
-
- if (missingResolvableAttrs.isEmpty) {
- val unresolvableAttrs = s.order.filterNot(_.resolved)
- logDebug(s"Failed to find $unresolvableAttrs in ${child.output.mkString(", ")}")
- s // Nothing we can do here. Return original plan.
- } else {
- // Add the missing attributes into projectList of Project/Window or
- // aggregateExpressions of Aggregate, if they are in the inputSet
- // but not in the outputSet of the plan.
- val newChild = child transformUp {
- case p: Project =>
- p.copy(projectList = p.projectList ++
- missingResolvableAttrs.filter((p.inputSet -- p.outputSet).contains))
- case w: Window =>
- w.copy(projectList = w.projectList ++
- missingResolvableAttrs.filter((w.inputSet -- w.outputSet).contains))
- case a: Aggregate =>
- val resolvableAttrs = missingResolvableAttrs.filter(a.groupingExpressions.contains)
- val notResolvedAttrs = resolvableAttrs.filterNot(a.aggregateExpressions.contains)
- val newAggregateExpressions = a.aggregateExpressions ++ notResolvedAttrs
- a.copy(aggregateExpressions = newAggregateExpressions)
- case o => o
- }
-
+ case s @ Sort(order, _, child) if !s.resolved && child.resolved =>
+ 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(newOrdering, s.global, newChild))
+ Sort(newOrder, s.global, addMissingAttr(child, missingAttrs)))
+ } else if (newOrder != order) {
+ s.copy(order = newOrder)
+ } else {
+ s
}
}
/**
- * Traverse the tree until resolving the sorting attributes
- * Return all the resolvable missing sorting attributes
- */
- @tailrec
- private def collectResolvableMissingAttrs(
- ordering: Seq[SortOrder],
- plan: LogicalPlan): (Seq[SortOrder], Seq[Attribute]) = {
+ * 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 plan
+ }
plan match {
- // Only Windows and Project have projectList-like attribute.
- case un: UnaryNode if un.isInstanceOf[Project] || un.isInstanceOf[Window] =>
- val (newOrdering, missingAttrs) = resolveAndFindMissing(ordering, un, un.child)
- // If missingAttrs is non empty, that means we got it and return it;
- // Otherwise, continue to traverse the tree.
- if (missingAttrs.nonEmpty) {
- (newOrdering, missingAttrs)
- } else {
- collectResolvableMissingAttrs(ordering, un.child)
- }
+ case p: Project =>
+ val missing = missingAttrs -- p.child.outputSet
+ Project(p.projectList ++ missingAttrs, addMissingAttr(p.child, missing))
+ case w: Window =>
+ val missing = missingAttrs -- w.child.outputSet
+ w.copy(projectList = w.projectList ++ missingAttrs,
+ child = addMissingAttr(w.child, missing))
case a: Aggregate =>
- val (newOrdering, missingAttrs) = resolveAndFindMissing(ordering, a, a.child)
- // For Aggregate, all the order by columns must be specified in group by clauses
- if (missingAttrs.nonEmpty &&
- missingAttrs.forall(ar => a.groupingExpressions.exists(_.semanticEquals(ar)))) {
- (newOrdering, missingAttrs)
- } else {
- // If missingAttrs is empty, we are unable to resolve any unresolved missing attributes
- (Seq.empty[SortOrder], Seq.empty[Attribute])
+ // all the missing attributes should be grouping expressions
+ // TODO: push down AggregateExpression
+ missingAttrs.foreach { attr =>
+ if (!a.groupingExpressions.contains(attr)) {
--- End diff --
I think we should use `a.groupingExprs.exists(_.semanticEquals(attr))` here, to be consistent with other places that do the similar checking like [CheckAnalysis](https://github.com/apache/spark/blob/master/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala#L157)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12705] [SQL] push missing attributes fo...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11153#issuecomment-182627104
**[Test build #2533 has finished](https://amplab.cs.berkeley.edu/jenkins/job/NewSparkPullRequestBuilder/2533/consoleFull)** for PR 11153 at commit [`bec639d`](https://github.com/apache/spark/commit/bec639de876e7b4dfccbc17a7bb94c212443e5ff).
* This patch **fails Spark unit tests**.
* This patch merges cleanly.
* This patch adds no public classes.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12705] [SQL] push missing attributes fo...
Posted by AmplabJenkins <gi...@git.apache.org>.
Github user AmplabJenkins commented on the pull request:
https://github.com/apache/spark/pull/11153#issuecomment-182753093
Test PASSed.
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/51085/
Test PASSed.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12705] [SQL] push missing attributes fo...
Posted by rxin <gi...@git.apache.org>.
Github user rxin commented on the pull request:
https://github.com/apache/spark/pull/11153#issuecomment-183228279
cc @cloud-fan for review too
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12705] [SQL] push missing attributes fo...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11153#issuecomment-182592913
**[Test build #2533 has started](https://amplab.cs.berkeley.edu/jenkins/job/NewSparkPullRequestBuilder/2533/consoleFull)** for PR 11153 at commit [`bec639d`](https://github.com/apache/spark/commit/bec639de876e7b4dfccbc17a7bb94c212443e5ff).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12705] [SQL] push missing attributes in...
Posted by SparkQA <gi...@git.apache.org>.
Github user SparkQA commented on the pull request:
https://github.com/apache/spark/pull/11153#issuecomment-182520156
**[Test build #51040 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/51040/consoleFull)** for PR 11153 at commit [`e1919cc`](https://github.com/apache/spark/commit/e1919ccf1e0a878bfb320d41beeffbc7f4c9b5da).
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] spark pull request: [SPARK-12705] [SQL] push missing attributes fo...
Posted by gatorsmile <gi...@git.apache.org>.
Github user gatorsmile commented on the pull request:
https://github.com/apache/spark/pull/11153#issuecomment-183197064
LGTM
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org