move to use a special case for Generate in ResolveReferrence

Author: Davies Liu

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala

@@ -378,6 +378,22 @@ class Analyzer(
         val newOrdering = resolveSortOrders(ordering, child, throws = false)
         Sort(newOrdering, global, child)
 
+      // A special case for Generate, because the output of Generate should not be resolved by
+      // ResolveReferences.
+      case g @ Generate(generator, join, outer, qualifier, output, child)
+        if child.resolved && !generator.resolved =>
+        val newG = generator transformUp {
+          case u @ UnresolvedAttribute(nameParts) =>
+            withPosition(u) { child.resolve(nameParts, resolver).getOrElse(u) }
+          case UnresolvedExtractValue(child, fieldExpr) =>
+            ExtractValue(child, fieldExpr, resolver)
+        }
+        if (newG.fastEquals(generator)) {
+          g
+        } else {
+          Generate(newG.asInstanceOf[Generator], join, outer, qualifier, output, child)
+        }
+
       case q: LogicalPlan =>
         logTrace(s"Attempting to resolve ${q.simpleString}")
         q transformExpressionsUp  {

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicOperators.scala

@@ -75,40 +75,6 @@ case class Generate(
   // as that will cause exceptions like unresolved attributes etc.
   override def expressions: Seq[Expression] = generator :: Nil
 
-  /**
-   * Runs [[transformUp]] with `rule` on all expressions present in this query operator.
-   * @param rule the rule to be applied to every expression in this operator.
-   *
-   * Note: This is used by ResolveReferences, we need to override it to not resolve
-   * `generatorOutput`.
-   */
-  override def transformExpressionsUp(rule: PartialFunction[Expression, Expression]): this.type = {
-    var changed = false
-
-    @inline def transformExpressionUp(e: Expression): Expression = {
-      val newE = e.transformUp(rule)
-      if (newE.fastEquals(e)) {
-        e
-      } else {
-        changed = true
-        newE
-      }
-    }
-
-    def recursiveTransform(arg: Any): AnyRef = arg match {
-      case e: Expression if expressions.contains(e) => transformExpressionUp(e)
-      case Some(e: Expression) if expressions.contains(e) => Some(transformExpressionUp(e))
-      case m: Map[_, _] => m
-      case d: DataType => d // Avoid unpacking Structs
-      case seq: Traversable[_] => seq.map(recursiveTransform)
-      case other: AnyRef => other
-    }
-
-    val newArgs = productIterator.map(recursiveTransform).toArray
-
-    if (changed) makeCopy(newArgs).asInstanceOf[this.type] else this
-  }
-
   def output: Seq[Attribute] = {
     val qualified = qualifier.map(q =>
       // prepend the new qualifier to the existed one