[SPARK-45584][SQL] Fix subquery execution failure with TakeOrderedAndProjectExec

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

@@ -283,6 +283,8 @@ case class TakeOrderedAndProjectExec(
   }
 
   override def executeCollect(): Array[InternalRow] = {
+    // SPARK-45584: Wait for subquery execution to finish before executing collect.
+    prepareAndWaitForSubqueries()
     val orderingSatisfies = SortOrder.orderingSatisfies(child.outputOrdering, sortOrder)
     val ord = new LazilyGeneratedOrdering(sortOrder, child.output)
     val limited = if (orderingSatisfies) {
@@ -299,6 +301,11 @@ case class TakeOrderedAndProjectExec(
     }
   }
 
+  private def prepareAndWaitForSubqueries(): Unit = {
+    prepare()
+    waitForSubqueries()
+  }
+
   private val serializer: Serializer = new UnsafeRowSerializer(child.output.size)
 
   private lazy val writeMetrics =

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

@@ -2711,4 +2711,28 @@ class SubquerySuite extends QueryTest
       checkAnswer(df, Row(1, "foo", 1, "foo"))
     }
   }
+
+  test("SPARK-45584: subquery execution should not fail with ORDER BY and LIMIT") {
+    withTable("t1") {
+      sql(
+        """
+          |CREATE TABLE t1 USING PARQUET
+          |AS SELECT * FROM VALUES
+          |(1, "a"),
+          |(2, "a"),
+          |(3, "a") t(id, value)
+          |""".stripMargin)
+      val df = sql(
+        """
+          |WITH t2 AS (
+          |  SELECT * FROM t1 ORDER BY id
+          |)
+          |SELECT *, (SELECT COUNT(*) FROM t2) FROM t2 LIMIT 10
+          |""".stripMargin)
+      // This should not fail with IllegalArgumentException.
+      checkAnswer(
+        df,
+        Row(1, "a", 3) :: Row(2, "a", 3) :: Row(3, "a", 3) :: Nil)
+    }
+  }
 }