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 infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
