(spark) branch branch-4.1 updated: [SPARK-56250][SQL] Remove confusing defensive code in SortExec.rowSorter and add warning comment

[vi...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

viirya pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/spark.git


The following commit(s) were added to refs/heads/branch-4.1 by this push:
     new d9cd19c6f193 [SPARK-56250][SQL] Remove confusing defensive code in SortExec.rowSorter and add warning comment
d9cd19c6f193 is described below

commit d9cd19c6f193d25d28bf27c86fbbde4c4f4cded1
Author: Liang-Chi Hsieh <vi...@gmail.com>
AuthorDate: Sat Mar 28 15:23:35 2026 -0700

    [SPARK-56250][SQL] Remove confusing defensive code in SortExec.rowSorter and add warning comment
    
    [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.
    
    To simplify the code and reduce confusion.
    
    No
    
    Existing tests
    
    Generated-by: Claude Sonnet 4.6
    
    Closes #55048 from viirya/revert-sort-exec-defensive-code.
    
    Authored-by: Liang-Chi Hsieh <vi...@gmail.com>
    Signed-off-by: Liang-Chi Hsieh <vi...@gmail.com>
    (cherry picked from commit 842d4dd09bea1f2ef4c0a819ade0151e501fd530)
    Signed-off-by: Liang-Chi Hsieh <vi...@gmail.com>
---
 .../org/apache/spark/sql/execution/SortExec.scala  | 26 +++++++++++-----------
 1 file changed, 13 insertions(+), 13 deletions(-)

diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/SortExec.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/SortExec.scala
index 11fde41aae9e..febfa0486927 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/SortExec.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/SortExec.scala
@@ -62,9 +62,10 @@ case class SortExec(
     "peakMemory" -> SQLMetrics.createSizeMetric(sparkContext, "peak memory"),
     "spillSize" -> SQLMetrics.createSizeMetric(sparkContext, "spill size"))
 
-  // Each task has its own instance of UnsafeExternalRowSorter. It is created in the
-  // createSorter method and stored in a ThreadLocal variable.
-  private[sql] var rowSorter: 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
@@ -99,14 +100,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] = {
@@ -196,14 +196,14 @@ case class SortExec(
   }
 
   /**
-   * In SortExec, we overwrites cleanupResources to close UnsafeExternalRowSorter.
+   * In SortExec, we overwrite cleanupResources to close UnsafeExternalRowSorter.
+   * There's possible for rowSorter to be null here, for example, in the scenario of empty iterator
+   * in the current task, the downstream physical node (like SortMergeJoinExec) will trigger
+   * cleanupResources before rowSorter is initialized in createSorter.
    */
   override protected[sql] def cleanupResources(): Unit = {
-    if (rowSorter != null && rowSorter.get() != null) {
-      // There's possible for rowSorter is null here, for example, in the scenario of empty
-      // iterator in the current task, the downstream physical node(like SortMergeJoinExec) will
-      // trigger cleanupResources before rowSorter initialized in createSorter.
-      rowSorter.get().cleanupResources()
+    if (rowSorter != null) {
+      rowSorter.cleanupResources()
     }
     super.cleanupResources()
   }


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org