rename to outputOrdering

Author: mgaido91

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala

@@ -740,8 +740,8 @@ object EliminateSorts extends Rule[LogicalPlan] {
  */
 object RemoveRedundantSorts extends Rule[LogicalPlan] {
   def apply(plan: LogicalPlan): LogicalPlan = plan transform {
-    case Sort(orders, true, child) if child.sortedOrder.nonEmpty
-        && child.sortedOrder.zip(orders).forall { case (s1, s2) => s1.satisfies(s2) } =>
+    case Sort(orders, true, child) if child.outputOrdering.nonEmpty
+        && SortOrder.orderingSatisfies(child.outputOrdering, orders) =>
       child
   }
 }

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

@@ -221,9 +221,9 @@ abstract class LogicalPlan
   def refresh(): Unit = children.foreach(_.refresh())
 
   /**
-   * If the current plan contains sorted data, it contains the sorted order.
+   * Returns the output ordering that this plan generates.
    */
-  def sortedOrder: Seq[SortOrder] = Nil
+  def outputOrdering: Seq[SortOrder] = Nil
 }
 
 /**
@@ -281,5 +281,5 @@ abstract class BinaryNode extends LogicalPlan {
 }
 
 abstract class KeepOrderUnaryNode extends UnaryNode {
-  override final def sortedOrder: Seq[SortOrder] = child.sortedOrder
+  override final def outputOrdering: Seq[SortOrder] = child.outputOrdering
 }

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

@@ -470,7 +470,7 @@ case class Sort(
     child: LogicalPlan) extends UnaryNode {
   override def output: Seq[Attribute] = child.output
   override def maxRows: Option[Long] = child.maxRows
-  override def sortedOrder: Seq[SortOrder] = order
+  override def outputOrdering: Seq[SortOrder] = order
 }
 
 /** Factory for constructing new `Range` nodes. */
@@ -525,7 +525,7 @@ case class Range(
     Statistics(sizeInBytes = LongType.defaultSize * numElements)
   }
 
-  override def sortedOrder: Seq[SortOrder] = output.map(a => SortOrder(a, Descending))
+  override def outputOrdering: Seq[SortOrder] = output.map(a => SortOrder(a, Descending))
 }
 
 case class Aggregate(
@@ -872,8 +872,8 @@ case class RepartitionByExpression(
   override def maxRows: Option[Long] = child.maxRows
   override def shuffle: Boolean = true
 
-  override def sortedOrder: Seq[SortOrder] = partitioning match {
-    case RangePartitioning(sortedOrder, _) => sortedOrder
+  override def outputOrdering: Seq[SortOrder] = partitioning match {
+    case RangePartitioning(ordering, _) => ordering
     case _ => Nil
   }
 }

sql/core/src/main/scala/org/apache/spark/sql/execution/ExistingRDD.scala

@@ -125,7 +125,7 @@ case class LogicalRDD(
     output: Seq[Attribute],
     rdd: RDD[InternalRow],
     outputPartitioning: Partitioning = UnknownPartitioning(0),
-    outputOrdering: Seq[SortOrder] = Nil,
+    override val outputOrdering: Seq[SortOrder] = Nil,
     override val isStreaming: Boolean = false)(session: SparkSession)
   extends LeafNode with MultiInstanceRelation {
 

sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryRelation.scala

@@ -170,5 +170,5 @@ case class InMemoryRelation(
   override protected def otherCopyArgs: Seq[AnyRef] =
     Seq(_cachedColumnBuffers, sizeInBytesStats, statsOfPlanToCache)
 
-  override def sortedOrder: Seq[SortOrder] = child.outputOrdering
+  override def outputOrdering: Seq[SortOrder] = child.outputOrdering
 }