Add fix

Author: MasterDDT

sql/core/src/main/scala/org/apache/spark/sql/execution/exchange/EnsureRequirements.scala

@@ -234,9 +234,15 @@ case class EnsureRequirements(conf: SQLConf) extends Rule[SparkPlan] {
 
     // Now that we've performed any necessary shuffles, add sorts to guarantee output orderings:
     children = children.zip(requiredChildOrderings).map { case (child, requiredOrdering) =>
+      logDebug(s"Checking if sort of ${requiredOrdering} needed on ${child.simpleString}")
       if (requiredOrdering.nonEmpty) {
         // If child.outputOrdering is [a, b] and requiredOrdering is [a], we do not need to sort.
-        if (requiredOrdering != child.outputOrdering.take(requiredOrdering.length)) {
+        val orderings = requiredOrdering zip child.outputOrdering
+        val needSort = orderings.length != requiredOrdering.length ||
+          orderings.exists { case (requiredOrder, childOrder) =>
+            !requiredOrder.semanticEquals(childOrder)
+        }
+        if (needSort) {
           SortExec(requiredOrdering, global = false, child = child)
         } else {
           child

sql/core/src/test/scala/org/apache/spark/sql/execution/PlannerSuite.scala

@@ -479,7 +479,7 @@ class PlannerSuite extends SharedSQLContext {
     }
   }
 
-  ignore("AIQ-2170 EnsureRequirements adds sort when ordering columns same but diff direction") {
+  test("EnsureRequirements adds sort when ordering columns same but diff direction") {
     val orderingA = SortOrder(Literal(1), Ascending)
     val orderingB = SortOrder(Literal(1), Descending)
     assert(orderingA != orderingB)
@@ -495,7 +495,7 @@ class PlannerSuite extends SharedSQLContext {
     }
   }
 
-  ignore("AIQ-2170 EnsureRequirements doesn't add sort with cached sorted table") {
+  test("EnsureRequirements doesn't add sort with cached sorted table") {
     withSQLConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {
       withTempTable("t1", "t2") {
         val df = Seq(
@@ -524,7 +524,7 @@ class PlannerSuite extends SharedSQLContext {
     }
   }
 
-  ignore("AIQ-2170 EnsureRequirements doesn't add sort with different column capitalization") {
+  test("EnsureRequirements doesn't add sort with different column capitalization") {
     withSQLConf(SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "-1") {
       withTempTable("t1", "t2") {
         val df = Seq(