Github user cloud-fan commented on a diff in the pull request:
https://github.com/apache/spark/pull/10678#discussion_r51660099
--- Diff:
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
---
@@ -521,38 +522,96 @@ class Analyzer(
*/
object ResolveSortReferences extends Rule[LogicalPlan] {
def apply(plan: LogicalPlan): LogicalPlan = plan resolveOperators {
- case s @ Sort(ordering, global, p @ Project(projectList, child))
- if !s.resolved && p.resolved =>
- val (newOrdering, missing) = resolveAndFindMissing(ordering, p,
child)
+ // Skip sort with aggregate. This will be handled in
ResolveAggregateFunctions
+ case sa @ Sort(_, _, child: Aggregate) => sa
- // If this rule was not a no-op, return the transformed plan,
otherwise return the original.
- if (missing.nonEmpty) {
- // Add missing attributes and then project them away after the
sort.
- Project(p.output,
- Sort(newOrdering, global,
- Project(projectList ++ missing, child)))
- } else {
- logDebug(s"Failed to find $missing in ${p.output.mkString(",
")}")
+ 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
+ }
+
+ // Add missing attributes and then project them away after the
sort.
+ Project(child.output,
+ Sort(newOrdering, s.global, newChild))
}
}
/**
- * Given a child and a grandchild that are present beneath a sort
operator, try to resolve
- * the sort ordering and returns it with a list of attributes that are
missing from the
- * child but are present in the grandchild.
+ * 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]) = {
+ 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 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])
+ }
+ // 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])
+ }
+ }
+
+ /**
+ * Try to resolve the sort ordering and returns it with a list of
attributes that are missing
+ * from the plan but are present in the child.
*/
- def resolveAndFindMissing(
+ private def resolveAndFindMissing(
ordering: Seq[SortOrder],
- child: LogicalPlan,
- grandchild: LogicalPlan): (Seq[SortOrder], Seq[Attribute]) = {
- val newOrdering = resolveSortOrders(ordering, grandchild, throws =
true)
+ plan: LogicalPlan,
+ child: LogicalPlan): (Seq[SortOrder], Seq[Attribute]) = {
+ val newOrdering = resolveSortOrders(ordering, child, throws = false)
--- End diff --
Now it won't throw exception for unresolved attribute here, so what will
happen if there are unresolved stuff?
---
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 [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
