[SPARK-56250][SQL] Remove confusing defensive code in SortExec.rowSorter and add warning comment

### What changes were proposed in this pull request?

[SPARK-52609](https://issues.apache.org/jira/browse/SPARK-52609) added some defensive code to SortExec. The defensive has some issues and was fixed by [SPARK-56203](https://issues.apache.org/jira/browse/SPARK-56203).

But actually the defensive code is not for Spark itself but to guard multithreading access to SortExec that isn't an issue to Spark itself. The defensive code could easily confuse others. After rethinking about it, it might be better to revert it and add some warning comment.

### Why are the changes needed?

To simplify the code and reduce confusion.

### Does this PR introduce _any_ user-facing change?

No

### How was this patch tested?

Existing tests

### Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Sonnet 4.6

Closes #55048 from viirya/revert-sort-exec-defensive-code.

Authored-by: Liang-Chi Hsieh <viirya@gmail.com>
Signed-off-by: Liang-Chi Hsieh <viirya@gmail.com>

Author: viirya

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

@@ -62,13 +62,10 @@ case class SortExec(
     "peakMemory" -> SQLMetrics.createSizeMetric(sparkContext, "peak memory"),
     "spillSize" -> SQLMetrics.createSizeMetric(sparkContext, "spill size"))
 
-  // Each task thread has its own UnsafeExternalRowSorter instance stored here.
-  // Using a stable lazy val (rather than a reassigned var) ensures that the ThreadLocal
-  // object itself is never replaced: concurrent tasks on different threads each get their
-  // own independent slot in the same ThreadLocal, so one task can never observe or clobber
-  // another task's sorter reference.
-  @transient private[sql] lazy val rowSorter: ThreadLocal[UnsafeExternalRowSorter] =
-    new ThreadLocal[UnsafeExternalRowSorter]()
+  // WARNING: This is a shared mutable var on the SortExec instance. Do not access it from
+  // multiple threads concurrently - Spark operators do not guarantee thread-safety and one
+  // task's sorter could overwrite another's, causing a race condition.
+  private[sql] var rowSorter: UnsafeExternalRowSorter = _
 
   /**
    * This method gets invoked only once for each SortExec instance to initialize an
@@ -101,14 +98,13 @@ case class SortExec(
     }
 
     val pageSize = SparkEnv.get.memoryManager.pageSizeBytes
-    val newRowSorter = UnsafeExternalRowSorter.create(
+    rowSorter = UnsafeExternalRowSorter.create(
       schema, ordering, prefixComparator, prefixComputer, pageSize, canUseRadixSort)
 
     if (testSpillFrequency > 0) {
-      newRowSorter.setTestSpillFrequency(testSpillFrequency)
+      rowSorter.setTestSpillFrequency(testSpillFrequency)
     }
-    rowSorter.set(newRowSorter)
-    rowSorter.get()
+    rowSorter
   }
 
   protected override def doExecute(): RDD[InternalRow] = {
@@ -204,8 +200,8 @@ case class SortExec(
    * cleanupResources before rowSorter is initialized in createSorter.
    */
   override protected[sql] def cleanupResources(): Unit = {
-    if (rowSorter.get() != null) {
-      rowSorter.get().cleanupResources()
+    if (rowSorter != null) {
+      rowSorter.cleanupResources()
     }
     super.cleanupResources()
   }