revert some changes

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

@@ -469,6 +469,70 @@ class Analyzer(override val catalogManager: CatalogManager)
     }
   }
 
+  /**
+   * Replaces [[UnresolvedAlias]]s with concrete aliases.
+   */
+  object ResolveAliases extends Rule[LogicalPlan] {
+    private def assignAliases(exprs: Seq[NamedExpression]) = {
+      def extractOnly(e: Expression): Boolean = e match {
+        case _: ExtractValue => e.children.forall(extractOnly)
+        case _: Literal => true
+        case _: Attribute => true
+        case _ => false
+      }
+      def metaForAutoGeneratedAlias = {
+        new MetadataBuilder()
+          .putString("__autoGeneratedAlias", "true")
+          .build()
+      }
+      exprs.map(_.transformUpWithPruning(_.containsPattern(UNRESOLVED_ALIAS)) {
+          case u @ UnresolvedAlias(child, optGenAliasFunc) =>
+          child match {
+            case ne: NamedExpression => ne
+            case go @ GeneratorOuter(g: Generator) if g.resolved => MultiAlias(go, Nil)
+            case e if !e.resolved => u
+            case g: Generator => MultiAlias(g, Nil)
+            case c @ Cast(ne: NamedExpression, _, _, _) => Alias(c, ne.name)()
+            case e: ExtractValue =>
+              if (extractOnly(e)) {
+                Alias(e, toPrettySQL(e))()
+              } else {
+                Alias(e, toPrettySQL(e))(explicitMetadata = Some(metaForAutoGeneratedAlias))
+              }
+            case e if optGenAliasFunc.isDefined =>
+              Alias(child, optGenAliasFunc.get.apply(e))()
+            case l: Literal => Alias(l, toPrettySQL(l))()
+            case e =>
+              Alias(e, toPrettySQL(e))(explicitMetadata = Some(metaForAutoGeneratedAlias))
+          }
+        }
+      ).asInstanceOf[Seq[NamedExpression]]
+    }
+
+    private def hasUnresolvedAlias(exprs: Seq[NamedExpression]) =
+      exprs.exists(_.exists(_.isInstanceOf[UnresolvedAlias]))
+
+    def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUpWithPruning(
+      _.containsPattern(UNRESOLVED_ALIAS), ruleId) {
+      case Aggregate(groups, aggs, child) if child.resolved && hasUnresolvedAlias(aggs) =>
+        Aggregate(groups, assignAliases(aggs), child)
+
+      case Pivot(groupByOpt, pivotColumn, pivotValues, aggregates, child)
+        if child.resolved && groupByOpt.isDefined && hasUnresolvedAlias(groupByOpt.get) =>
+        Pivot(Some(assignAliases(groupByOpt.get)), pivotColumn, pivotValues, aggregates, child)
+
+      case up: Unpivot if up.child.resolved &&
+        (up.ids.exists(hasUnresolvedAlias) || up.values.exists(_.exists(hasUnresolvedAlias))) =>
+        up.copy(ids = up.ids.map(assignAliases), values = up.values.map(_.map(assignAliases)))
+
+      case Project(projectList, child) if child.resolved && hasUnresolvedAlias(projectList) =>
+        Project(assignAliases(projectList), child)
+
+      case c: CollectMetrics if c.child.resolved && hasUnresolvedAlias(c.metrics) =>
+        c.copy(metrics = assignAliases(c.metrics))
+    }
+  }
+
   object ResolveGroupingAnalytics extends Rule[LogicalPlan] {
     private[analysis] def hasGroupingFunction(e: Expression): Boolean = {
       e.exists (g => g.isInstanceOf[Grouping] || g.isInstanceOf[GroupingID])
@@ -4209,67 +4273,3 @@ object RemoveTempResolvedColumn extends Rule[LogicalPlan] {
   }
 }
 
-/**
- * Replaces [[UnresolvedAlias]]s with concrete aliases.
- */
-object ResolveAliases extends Rule[LogicalPlan] {
-  def metaForAutoGeneratedAlias: Metadata = {
-    new MetadataBuilder()
-      .putString("__autoGeneratedAlias", "true")
-      .build()
-  }
-
-  def assignAliases(exprs: Seq[NamedExpression]): Seq[NamedExpression] = {
-    def extractOnly(e: Expression): Boolean = e match {
-      case _: ExtractValue => e.children.forall(extractOnly)
-      case _: Literal => true
-      case _: Attribute => true
-      case _ => false
-    }
-    exprs.map(_.transformUpWithPruning(_.containsPattern(UNRESOLVED_ALIAS)) {
-      case u @ UnresolvedAlias(child, optGenAliasFunc) =>
-        child match {
-          case ne: NamedExpression => ne
-          case go @ GeneratorOuter(g: Generator) if g.resolved => MultiAlias(go, Nil)
-          case e if !e.resolved => u
-          case g: Generator => MultiAlias(g, Nil)
-          case c @ Cast(ne: NamedExpression, _, _, _) => Alias(c, ne.name)()
-          case e: ExtractValue =>
-            if (extractOnly(e)) {
-              Alias(e, toPrettySQL(e))()
-            } else {
-              Alias(e, toPrettySQL(e))(explicitMetadata = Some(metaForAutoGeneratedAlias))
-            }
-          case e if optGenAliasFunc.isDefined =>
-            Alias(child, optGenAliasFunc.get.apply(e))()
-          case l: Literal => Alias(l, toPrettySQL(l))()
-          case e =>
-            Alias(e, toPrettySQL(e))(explicitMetadata = Some(metaForAutoGeneratedAlias))
-        }
-    }
-    ).asInstanceOf[Seq[NamedExpression]]
-  }
-
-  private def hasUnresolvedAlias(exprs: Seq[NamedExpression]) =
-    exprs.exists(_.exists(_.isInstanceOf[UnresolvedAlias]))
-
-  def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsUpWithPruning(
-    _.containsPattern(UNRESOLVED_ALIAS), ruleId) {
-    case Aggregate(groups, aggs, child) if child.resolved && hasUnresolvedAlias(aggs) =>
-      Aggregate(groups, assignAliases(aggs), child)
-
-    case Pivot(groupByOpt, pivotColumn, pivotValues, aggregates, child)
-      if child.resolved && groupByOpt.isDefined && hasUnresolvedAlias(groupByOpt.get) =>
-      Pivot(Some(assignAliases(groupByOpt.get)), pivotColumn, pivotValues, aggregates, child)
-
-    case up: Unpivot if up.child.resolved &&
-      (up.ids.exists(hasUnresolvedAlias) || up.values.exists(_.exists(hasUnresolvedAlias))) =>
-      up.copy(ids = up.ids.map(assignAliases), values = up.values.map(_.map(assignAliases)))
-
-    case Project(projectList, child) if child.resolved && hasUnresolvedAlias(projectList) =>
-      Project(assignAliases(projectList), child)
-
-    case c: CollectMetrics if c.child.resolved && hasUnresolvedAlias(c.metrics) =>
-      c.copy(metrics = assignAliases(c.metrics))
-  }
-}

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

@@ -446,7 +446,6 @@ trait CheckAnalysis extends PredicateHelper with LookupCatalog with QueryErrorsB
 
             groupingExprs.foreach(checkValidGroupingExprs)
             aggregateExprs.foreach(checkValidAggregateExpression)
-            // TODO: if the Aggregate is resolved, it can't contain the LateralColumnAliasReference
 
           case CollectMetrics(name, metrics, _) =>
             if (name == null || name.isEmpty) {

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

@@ -23,6 +23,7 @@ import org.apache.spark.sql.catalyst.plans.logical.{Aggregate, LogicalPlan, Proj
 import org.apache.spark.sql.catalyst.rules.Rule
 import org.apache.spark.sql.catalyst.trees.TreeNodeTag
 import org.apache.spark.sql.catalyst.trees.TreePattern.LATERAL_COLUMN_ALIAS_REFERENCE
+import org.apache.spark.sql.catalyst.util.toPrettySQL
 import org.apache.spark.sql.errors.QueryCompilationErrors
 import org.apache.spark.sql.internal.SQLConf
 
@@ -101,6 +102,14 @@ object ResolveLateralColumnAliasReference extends Rule[LogicalPlan] {
    */
   val NAME_PARTS_FROM_UNRESOLVED_ATTR = TreeNodeTag[Seq[String]]("name_parts_from_unresolved_attr")
 
+  private def assignAlias(expr: Expression): NamedExpression = {
+    expr match {
+      case ne: NamedExpression => ne
+      case e =>
+        Alias(e, toPrettySQL(e))()
+    }
+  }
+
   override def apply(plan: LogicalPlan): LogicalPlan = {
     if (!conf.getConf(SQLConf.LATERAL_COLUMN_ALIAS_IMPLICIT_ENABLED)) {
       plan
@@ -172,24 +181,15 @@ object ResolveLateralColumnAliasReference extends Rule[LogicalPlan] {
                         lcaRef.nameParts, aggExpr)
                   }
                 }
-                val ne = expressionMap.getOrElseUpdate(
-                  aggExpr.canonicalized,
-                  ResolveAliases.assignAliases(Seq(UnresolvedAlias(aggExpr))).map {
-                    // TODO temporarily clear the metadata for an issue found in test
-                    case a: Alias => a.copy(a.child, a.name)(
-                      a.exprId, a.qualifier, None, a.nonInheritableMetadataKeys)
-                    case other => other
-                  }.head)
+                val ne = expressionMap.getOrElseUpdate(aggExpr.canonicalized, assignAlias(aggExpr))
                 newAggExprs += ne
                 ne.toAttribute
               case e if groupingExpressions.exists(_.semanticEquals(e)) =>
                 // TODO one concern here, is condition here be able to match all grouping
                 //  expressions? For example, Agg [age + 10] [1 + age + 10], when transforming down,
                 //  is it possible that (1 + age) + 10, so that it won't be able to match (age + 10)
                 //  add a test.
-                val ne = expressionMap.getOrElseUpdate(
-                  e.canonicalized,
-                  ResolveAliases.assignAliases(Seq(UnresolvedAlias(e))).head)
+                val ne = expressionMap.getOrElseUpdate(e.canonicalized, assignAlias(e))
                 newAggExprs += ne
                 ne.toAttribute
             }.asInstanceOf[NamedExpression]

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleIdCollection.scala

@@ -49,6 +49,7 @@ object RuleIdCollection {
       "org.apache.spark.sql.catalyst.analysis.Analyzer$GlobalAggregates" ::
       "org.apache.spark.sql.catalyst.analysis.Analyzer$ResolveAggAliasInGroupBy" ::
       "org.apache.spark.sql.catalyst.analysis.Analyzer$ResolveAggregateFunctions" ::
+      "org.apache.spark.sql.catalyst.analysis.Analyzer$ResolveAliases" ::
       "org.apache.spark.sql.catalyst.analysis.Analyzer$ResolveBinaryArithmetic" ::
       "org.apache.spark.sql.catalyst.analysis.Analyzer$ResolveDeserializer" ::
       "org.apache.spark.sql.catalyst.analysis.Analyzer$ResolveEncodersInUDF" ::
@@ -82,7 +83,6 @@ object RuleIdCollection {
       "org.apache.spark.sql.catalyst.analysis.DeduplicateRelations" ::
       "org.apache.spark.sql.catalyst.analysis.EliminateSubqueryAliases" ::
       "org.apache.spark.sql.catalyst.analysis.EliminateUnions" ::
-      "org.apache.spark.sql.catalyst.analysis.ResolveAliases" ::
       "org.apache.spark.sql.catalyst.analysis.ResolveDefaultColumns" ::
       "org.apache.spark.sql.catalyst.analysis.ResolveExpressionsWithNamePlaceholders" ::
       "org.apache.spark.sql.catalyst.analysis.ResolveHints$ResolveCoalesceHints" ::