[SPARK-51796] [SQL] Disallow Sort order expressions under non-Sort operators

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

@@ -823,6 +823,7 @@ trait CheckAnalysis extends LookupCatalog with QueryErrorsBase with PlanToString
 
           case p @ Project(projectList, _) =>
             checkForUnspecifiedWindow(projectList)
+            checkExpressionUnderUnsupportedOperator(p)
 
           case agg@Aggregate(_, aggregateExpressions, _, _) if
             PlanHelper.specialExpressionsInUnsupportedOperator(agg).isEmpty =>
@@ -932,12 +933,7 @@ trait CheckAnalysis extends LookupCatalog with QueryErrorsBase with PlanToString
                 "expressionList" -> invalidExprSqls.mkString(", ")))
 
           case other if PlanHelper.specialExpressionsInUnsupportedOperator(other).nonEmpty =>
-            val invalidExprSqls =
-              PlanHelper.specialExpressionsInUnsupportedOperator(other).map(toSQLExpr)
-            other.failAnalysis(
-              errorClass = "UNSUPPORTED_EXPR_FOR_OPERATOR",
-              messageParameters = Map(
-                "invalidExprSqls" -> invalidExprSqls.mkString(", ")))
+            checkExpressionUnderUnsupportedOperator(other)
 
           case _ => // Analysis successful!
         }
@@ -1226,6 +1222,17 @@ trait CheckAnalysis extends LookupCatalog with QueryErrorsBase with PlanToString
       }
     }
   }
+
+  private def checkExpressionUnderUnsupportedOperator(plan: LogicalPlan): Unit = {
+    if (PlanHelper.specialExpressionsInUnsupportedOperator(plan).nonEmpty) {
+      val invalidExprSqls =
+        PlanHelper.specialExpressionsInUnsupportedOperator(plan).map(toSQLExpr)
+      plan.failAnalysis(
+        errorClass = "UNSUPPORTED_EXPR_FOR_OPERATOR",
+        messageParameters = Map(
+          "invalidExprSqls" -> invalidExprSqls.mkString(", ")))
+    }
+  }
 }
 
 // a heap of the preempted error that only keeps the top priority element, representing the sole

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

@@ -17,7 +17,7 @@
 
 package org.apache.spark.sql.catalyst.plans.logical
 
-import org.apache.spark.sql.catalyst.expressions.{Expression, Generator, WindowExpression}
+import org.apache.spark.sql.catalyst.expressions.{Expression, Generator, SortOrder, WindowExpression}
 import org.apache.spark.sql.catalyst.expressions.aggregate.AggregateExpression
 
 /**
@@ -31,6 +31,7 @@ object PlanHelper {
    * - A WindowExpression but the plan is not Window
    * - An AggregateExpression but the plan is not Aggregate or Window
    * - A Generator but the plan is not Generate
+   * - A SortOrder but the plan is not Sort
    * Returns the list of invalid expressions that this operator hosts. This can happen when
    * 1. The input query from users contain invalid expressions.
    *    Example : SELECT * FROM tab WHERE max(c1) > 0
@@ -50,6 +51,8 @@ object PlanHelper {
         case e: Generator
           if !(plan.isInstanceOf[Generate] ||
                plan.isInstanceOf[BaseEvalPythonUDTF]) => e
+        case sortOrder: SortOrder if !plan.isInstanceOf[Sort] && !plan.isInstanceOf[Window] =>
+          sortOrder
       }
     }
     invalidExpressions

sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala

@@ -4934,6 +4934,17 @@ class SQLQuerySuite extends QueryTest with SharedSparkSession with AdaptiveSpark
 
     checkAnswer(df, Row("a"))
   }
+
+  test("SPARK-51796: SortOrder expressions are not allowed under non-Sort operators") {
+    val df = sql("SELECT 1 AS name")
+    checkError(
+      exception = intercept[AnalysisException] {
+        df.select(desc("name"))
+      },
+      condition = "UNSUPPORTED_EXPR_FOR_OPERATOR",
+      parameters = Map("invalidExprSqls" -> "\"name DESC NULLS LAST\""),
+    )
+  }
 }
 
 case class Foo(bar: Option[String])