[SPARK-48006][SQL]add SortOrder for window function which has no orde…

…rSpec

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala

@@ -2434,6 +2434,13 @@ class AstBuilder extends DataTypeAstBuilder with SQLConfHelper with Logging {
     val partition = ctx.partition.asScala.map(expression)
     val order = ctx.sortItem.asScala.map(visitSortItem)
 
+    // Add SortOrder for window expresssion
+    if (ctx.sortItem.asScala.isEmpty && conf.hiveWindowFunctionOrderSpec) {
+      order = ctx.partition.asScala.map { expr =>
+        SortOrder(expression(expr), Ascending, Ascending.defaultNullOrdering, Seq.empty)
+      }
+    }
+
     // RANGE/ROWS BETWEEN ...
     val frameSpecOption = Option(ctx.windowFrame).map { frame =>
       val frameType = frame.frameType.getType match {

sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala

@@ -2564,6 +2564,18 @@ object SQLConf {
       .intConf
       .createWithDefault(10)
 
+  val HIVE_WINDOW_FUNCTION_ORDER_SPEC =
+    buildConf("spark.sql.hive.window.function.orderSpec")
+      .internal()
+      .doc("Whether to add order expression for window function.The parser requires window to be" +
+        " ordered otherwise throw analyzed exception. When true, it will add order partition " +
+        "for  window function which has no orderSpec.")
+      .version("4.0.0")
+      .booleanConf
+      .createWithDefault(false)
+
+
+
   val FILE_SOURCE_LOG_CLEANUP_DELAY =
     buildConf("spark.sql.streaming.fileSource.log.cleanupDelay")
       .internal()
@@ -5044,6 +5056,8 @@ class SQLConf extends Serializable with Logging with SqlApiConf {
 
   def fileSourceLogCompactInterval: Int = getConf(FILE_SOURCE_LOG_COMPACT_INTERVAL)
 
+  def hiveWindowFunctionOrderSpec: Boolean = getConf(HIVE_WINDOW_FUNCTION_ORDER_SPEC)
+
   def fileSourceLogCleanupDelay: Long = getConf(FILE_SOURCE_LOG_CLEANUP_DELAY)
 
   def streamingSchemaInference: Boolean = getConf(STREAMING_SCHEMA_INFERENCE)