Update code

Author: beliefer

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/V2ScanRelationPushDown.scala

@@ -257,11 +257,15 @@ object V2ScanRelationPushDown extends Rule[LogicalPlan] with PredicateHelper {
     case s @ Sort(order, _, operation @ ScanOperation(_, filter, sHolder: ScanBuilderHolder))
       if filter.isEmpty =>
       val orders = DataSourceStrategy.translateSortOrders(order)
-      val topNPushed = PushDownUtils.pushTopN(sHolder.builder, orders.toArray, limit)
-      if (topNPushed) {
-        sHolder.pushedLimit = Some(limit)
-        sHolder.sortValues = orders
-        operation
+      if (orders.length == order.length) {
+        val topNPushed = PushDownUtils.pushTopN(sHolder.builder, orders.toArray, limit)
+        if (topNPushed) {
+          sHolder.pushedLimit = Some(limit)
+          sHolder.sortValues = orders
+          operation
+        } else {
+          s
+        }
       } else {
         s
       }

sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCV2Suite.scala

@@ -227,6 +227,13 @@ class JDBCV2Suite extends QueryTest with SharedSparkSession with ExplainSuiteHel
     // LIMIT is pushed down only if all the filters are pushed down
     checkPushedLimit(df7)
     checkAnswer(df7, Seq(Row(10000.00, 1000.0, "amy")))
+
+    val df8 = spark.read
+      .table("h2.test.employee")
+      .sort(sub($"NAME"))
+      .limit(1)
+    checkPushedLimit(df8)
+    checkAnswer(df8, Seq(Row(2, "alex", 12000.00, 1200.0)))
   }
 
   private def createSortValues(