Revert accidental metadata changes

Signed-off-by: Karen Feng <[email protected]>

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

@@ -61,6 +61,4 @@ case class EventTimeWatermark(
       a
     }
   }
-
-  override val metadataOutput: Seq[Attribute] = child.metadataOutput
 }

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

@@ -41,7 +41,6 @@ import org.apache.spark.util.random.RandomSampler
 case class ReturnAnswer(child: LogicalPlan) extends UnaryNode {
   override def maxRows: Option[Long] = child.maxRows
   override def output: Seq[Attribute] = child.output
-  override def metadataOutput: Seq[Attribute] = child.metadataOutput
 }
 
 /**
@@ -53,7 +52,6 @@ case class ReturnAnswer(child: LogicalPlan) extends UnaryNode {
  */
 case class Subquery(child: LogicalPlan, correlated: Boolean) extends OrderPreservingUnaryNode {
   override def output: Seq[Attribute] = child.output
-  override def metadataOutput: Seq[Attribute] = child.metadataOutput
 }
 
 object Subquery {
@@ -118,8 +116,6 @@ case class Generate(
     child: LogicalPlan)
   extends UnaryNode {
 
-  val unrequiredSet: Set[Int] = unrequiredChildIndex.toSet
-
   lazy val requiredChildOutput: Seq[Attribute] = {
     val unrequiredSet = unrequiredChildIndex.toSet
     child.output.zipWithIndex.filterNot(t => unrequiredSet.contains(t._2)).map(_._1)
@@ -147,13 +143,11 @@ case class Generate(
   }
 
   def output: Seq[Attribute] = requiredChildOutput ++ qualifiedGeneratorOutput
-  override def metadataOutput: Seq[Attribute] = child.metadataOutput
 }
 
 case class Filter(condition: Expression, child: LogicalPlan)
   extends OrderPreservingUnaryNode with PredicateHelper {
   override def output: Seq[Attribute] = child.output
-  override def metadataOutput: Seq[Attribute] = child.metadataOutput
 
   override def maxRows: Option[Long] = child.maxRows
 
@@ -223,7 +217,6 @@ case class Except(
   override def nodeName: String = getClass.getSimpleName + ( if ( isAll ) "All" else "" )
   /** We don't use right.output because those rows get excluded from the set. */
   override def output: Seq[Attribute] = left.output
-  override def metadataOutput: Seq[Attribute] = left.metadataOutput
 
   override def metadataOutput: Seq[Attribute] = Nil
 
@@ -556,8 +549,6 @@ object View {
 case class With(child: LogicalPlan, cteRelations: Seq[(String, SubqueryAlias)]) extends UnaryNode {
   override def output: Seq[Attribute] = child.output
 
-  override def metadataOutput: Seq[Attribute] = child.metadataOutput
-
   override def simpleString(maxFields: Int): String = {
     val cteAliases = truncatedString(cteRelations.map(_._1), "[", ", ", "]", maxFields)
     s"CTE $cteAliases"
@@ -570,7 +561,6 @@ case class WithWindowDefinition(
     windowDefinitions: Map[String, WindowSpecDefinition],
     child: LogicalPlan) extends UnaryNode {
   override def output: Seq[Attribute] = child.output
-  override def metadataOutput: Seq[Attribute] = child.metadataOutput
 }
 
 /**
@@ -584,7 +574,6 @@ case class Sort(
     global: Boolean,
     child: LogicalPlan) extends UnaryNode {
   override def output: Seq[Attribute] = child.output
-  override def metadataOutput: Seq[Attribute] = child.metadataOutput
   override def maxRows: Option[Long] = child.maxRows
   override def outputOrdering: Seq[SortOrder] = order
 }
@@ -710,7 +699,6 @@ case class Window(
   override def maxRows: Option[Long] = child.maxRows
   override def output: Seq[Attribute] =
     child.output ++ windowExpressions.map(_.toAttribute)
-  override def metadataOutput: Seq[Attribute] = child.metadataOutput
 
   override def producedAttributes: AttributeSet = windowOutputSet
 
@@ -906,7 +894,6 @@ object Limit {
  */
 case class GlobalLimit(limitExpr: Expression, child: LogicalPlan) extends OrderPreservingUnaryNode {
   override def output: Seq[Attribute] = child.output
-  override def metadataOutput: Seq[Attribute] = child.metadataOutput
   override def maxRows: Option[Long] = {
     limitExpr match {
       case IntegerLiteral(limit) => Some(limit)
@@ -923,7 +910,6 @@ case class GlobalLimit(limitExpr: Expression, child: LogicalPlan) extends OrderP
  */
 case class LocalLimit(limitExpr: Expression, child: LogicalPlan) extends OrderPreservingUnaryNode {
   override def output: Seq[Attribute] = child.output
-  override def metadataOutput: Seq[Attribute] = child.metadataOutput
 
   override def maxRowsPerPartition: Option[Long] = {
     limitExpr match {
@@ -945,7 +931,6 @@ case class LocalLimit(limitExpr: Expression, child: LogicalPlan) extends OrderPr
  */
 case class Tail(limitExpr: Expression, child: LogicalPlan) extends OrderPreservingUnaryNode {
   override def output: Seq[Attribute] = child.output
-  override def metadataOutput: Seq[Attribute] = child.metadataOutput
   override def maxRows: Option[Long] = {
     limitExpr match {
       case IntegerLiteral(limit) => Some(limit)
@@ -1031,7 +1016,6 @@ case class Sample(
 
   override def maxRows: Option[Long] = child.maxRows
   override def output: Seq[Attribute] = child.output
-  override def metadataOutput: Seq[Attribute] = child.metadataOutput
 }
 
 /**
@@ -1040,7 +1024,6 @@ case class Sample(
 case class Distinct(child: LogicalPlan) extends UnaryNode {
   override def maxRows: Option[Long] = child.maxRows
   override def output: Seq[Attribute] = child.output
-  override def metadataOutput: Seq[Attribute] = child.metadataOutput
 }
 
 /**
@@ -1145,7 +1128,6 @@ case class Deduplicate(
     child: LogicalPlan) extends UnaryNode {
   override def maxRows: Option[Long] = child.maxRows
   override def output: Seq[Attribute] = child.output
-  override def metadataOutput: Seq[Attribute] = child.metadataOutput
 }
 
 /**
@@ -1174,5 +1156,4 @@ case class CollectMetrics(
   }
 
   override def output: Seq[Attribute] = child.output
-  override def metadataOutput: Seq[Attribute] = child.metadataOutput
 }

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

@@ -31,7 +31,6 @@ case class UnresolvedHint(name: String, parameters: Seq[Any], child: LogicalPlan
 
   override lazy val resolved: Boolean = false
   override def output: Seq[Attribute] = child.output
-  override def metadataOutput: Seq[Attribute] = child.metadataOutput
 }
 
 /**
@@ -43,8 +42,6 @@ case class ResolvedHint(child: LogicalPlan, hints: HintInfo = HintInfo())
 
   override def output: Seq[Attribute] = child.output
 
-  override def metadataOutput: Seq[Attribute] = child.metadataOutput
-
   override def doCanonicalize(): LogicalPlan = child.canonicalized
 }
 

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

@@ -238,8 +238,6 @@ case class TypedFilter(
 
   override def output: Seq[Attribute] = child.output
 
-  override def metadataOutput: Seq[Attribute] = child.metadataOutput
-
   def withObjectProducerChild(obj: LogicalPlan): Filter = {
     assert(obj.output.length == 1)
     Filter(typedCondition(obj.output.head), obj)
@@ -335,8 +333,6 @@ case class AppendColumns(
 
   override def output: Seq[Attribute] = child.output ++ newColumns
 
-  override def metadataOutput: Seq[Attribute] = child.metadataOutput
-
   def newColumns: Seq[Attribute] = serializer.map(_.toAttribute)
 }
 
@@ -350,8 +346,6 @@ case class AppendColumnsWithObject(
     child: LogicalPlan) extends ObjectConsumer {
 
   override def output: Seq[Attribute] = (childSerializer ++ newColumnsSerializer).map(_.toAttribute)
-
-  override def metadataOutput: Seq[Attribute] = child.metadataOutput
 }
 
 /** Factory for constructing new `MapGroups` nodes. */

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

@@ -80,8 +80,6 @@ trait BaseEvalPython extends UnaryNode {
 
   override def output: Seq[Attribute] = child.output ++ resultAttrs
 
-  override def metadataOutput: Seq[Attribute] = child.metadataOutput
-
   override def producedAttributes: AttributeSet = AttributeSet(resultAttrs)
 }