[auto-merge] release/26.08 to main [skip ci] [bot] (NVIDIA#15483)

auto-merge triggered by github actions on `release/26.08` to create a PR
keeping `main` up-to-date. If this PR is unable to be merged due to
conflicts, it will remain open until manually fix.

Author: nvauto

sql-plugin/src/main/scala/com/nvidia/spark/rapids/RowBasedShuffleChecksumConf.scala

@@ -0,0 +1,50 @@
+/*
+ * Copyright (c) 2026, NVIDIA CORPORATION.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.nvidia.spark.rapids
+
+import scala.util.Try
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.internal.SQLConf
+
+object RowBasedShuffleChecksumConf {
+  val ChecksumEnabledKey = "spark.sql.shuffle.orderIndependentChecksum.enabled"
+  val ChecksumMismatchFullRetryKey =
+    "spark.sql.shuffle.orderIndependentChecksum.enableFullRetryOnMismatch"
+
+  // SQLConf takes priority over SparkConf when explicitly set (e.g. SET command mid-session).
+  // SparkConf is checked next for values set at session start or via --conf.
+  // If neither is explicitly set, we fall back to Spark's registered config default via
+  // SQLConf.getConfString: on Spark 4.2+ both keys default to true (checksums on by
+  // default), so RAPIDS will fall back to SortShuffleManager unless the user explicitly
+  // disables them (set both keys to false). On Spark < 4.2 these keys default to false
+  // or are not registered in SQLConf, so GPU shuffle proceeds normally.
+  def isEnabled(sqlConf: SQLConf, sparkConf: SparkConf): Boolean = {
+    getBoolean(sqlConf, sparkConf, ChecksumEnabledKey) ||
+      getBoolean(sqlConf, sparkConf, ChecksumMismatchFullRetryKey)
+  }
+
+  private def getBoolean(sqlConf: SQLConf, sparkConf: SparkConf, key: String): Boolean = {
+    if (sqlConf.contains(key)) {
+      sqlConf.getConfString(key).toBoolean
+    } else if (sparkConf.contains(key)) {
+      sparkConf.getBoolean(key, false)
+    } else {
+      Try(sqlConf.getConfString(key)).map(_.toBoolean).getOrElse(false)
+    }
+  }
+}

sql-plugin/src/main/scala/org/apache/spark/sql/rapids/GpuShuffleDependency.scala

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2020-2022, NVIDIA CORPORATION.
+ * Copyright (c) 2020-2026, NVIDIA CORPORATION.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -40,5 +40,10 @@ class GpuShuffleDependency[K: ClassTag, V: ClassTag, C: ClassTag](
   extends ShuffleDependency[K, V, C](rdd, partitioner, serializer, keyOrdering,
     aggregator, mapSideCombine, shuffleWriterProcessor) {
 
+  // Set by RapidsShuffleInternalManagerBase.registerShuffle to record the
+  // registration-time checksum fallback decision so writer and reader use
+  // the same path even if SQLConf changes between calls.
+  var checksumFallback: Boolean = false
+
   override def toString: String = "GPU Shuffle Dependency"
 }

sql-plugin/src/main/scala/org/apache/spark/sql/rapids/GpuShuffleEnv.scala

@@ -19,10 +19,12 @@ package org.apache.spark.sql.rapids
 import java.util.Locale
 
 import com.nvidia.spark.rapids._
+import com.nvidia.spark.rapids.RowBasedShuffleChecksumConf
 import com.nvidia.spark.rapids.shims.ShuffleManagerShimUtils
 
 import org.apache.spark.{SparkConf, SparkEnv}
 import org.apache.spark.internal.Logging
+import org.apache.spark.sql.internal.SQLConf
 
 class GpuShuffleEnv(rapidsConf: RapidsConf) extends Logging {
   private var shuffleCatalog: ShuffleBufferCatalog = _
@@ -144,10 +146,7 @@ object GpuShuffleEnv extends Logging {
   // which is for IO-level corruption diagnosis and IS supported by RAPIDS shuffle.
   def isRowBasedChecksumEnabled: Boolean = {
     val conf = SparkEnv.get.conf
-    conf.getBoolean(
-      "spark.sql.shuffle.orderIndependentChecksum.enabled", false) ||
-    conf.getBoolean(
-      "spark.sql.shuffle.orderIndependentChecksum.enableFullRetryOnMismatch", false)
+    RowBasedShuffleChecksumConf.isEnabled(SQLConf.get, conf)
   }
 
   //

sql-plugin/src/main/scala/org/apache/spark/sql/rapids/RapidsShuffleInternalManagerBase.scala

@@ -1736,7 +1736,9 @@ class RapidsShuffleInternalManagerBase(conf: SparkConf, val isDriver: Boolean)
   // NOTE: this can be null in the driver side.
   protected lazy val env = SparkEnv.get
   protected lazy val blockManager = env.blockManager
-  protected lazy val shouldFallThroughOnEverything = {
+  // Stable reasons to always fall back to SortShuffleManager, evaluated once at
+  // first shuffle registration.
+  protected lazy val shouldAlwaysFallBack = {
     val fallThroughReasons = new ListBuffer[String]()
     if (!rapidsConf.isMultiThreadedShuffleManagerMode) {
       if (GpuShuffleEnv.isExternalShuffleEnabled) {
@@ -1749,19 +1751,28 @@ class RapidsShuffleInternalManagerBase(conf: SparkConf, val isDriver: Boolean)
     if (rapidsConf.isSqlExplainOnlyEnabled) {
       fallThroughReasons += "Plugin is in explain only mode"
     }
-    if (GpuShuffleEnv.isRowBasedChecksumEnabled) {
-      fallThroughReasons += "Detected order-independent checksum enabled " +
-        "(spark.sql.shuffle.orderIndependentChecksum.enabled or " +
-        "enableFullRetryOnMismatch). " +
-        "This Spark 4.1+ feature is not yet supported by Spark-Rapids."
-    }
     if (fallThroughReasons.nonEmpty) {
       logWarning(s"Rapids Shuffle Plugin is falling back to SortShuffleManager " +
         s"because: ${fallThroughReasons.mkString(", ")}")
     }
     fallThroughReasons.nonEmpty
   }
 
+  private val rowBasedChecksumFallbackLogged = new AtomicBoolean(false)
+
+  private def shouldFallThroughForShuffle: Boolean = {
+    val rowBasedChecksumFallback = GpuShuffleEnv.isRowBasedChecksumEnabled
+    if (rowBasedChecksumFallback) {
+      if (rowBasedChecksumFallbackLogged.compareAndSet(false, true)) {
+        logWarning("Rapids Shuffle Plugin is falling back to SortShuffleManager because: " +
+          "Detected order-independent checksum enabled " +
+          "(spark.sql.shuffle.orderIndependentChecksum.enabled or enableFullRetryOnMismatch). " +
+          "This Spark 4.1+ feature is not yet supported by Spark-Rapids.")
+      }
+    }
+    shouldAlwaysFallBack || rowBasedChecksumFallback
+  }
+
   private lazy val localBlockManagerId = blockManager.blockManagerId
 
   // Used to prevent stopping multiple times RAPIDS Shuffle Manager internals.
@@ -1776,7 +1787,7 @@ class RapidsShuffleInternalManagerBase(conf: SparkConf, val isDriver: Boolean)
         "RapidsShuffleManager is configured"))
 
   protected lazy val resolver =
-    if (shouldFallThroughOnEverything) {
+    if (shouldAlwaysFallBack) {
       wrapped.shuffleBlockResolver
     } else if (rapidsConf.isMultiThreadedShuffleManagerMode) {
       // MULTITHREADED mode: use GpuShuffleBlockResolver
@@ -1848,11 +1859,13 @@ class RapidsShuffleInternalManagerBase(conf: SparkConf, val isDriver: Boolean)
     val orig = wrapped.registerShuffle(shuffleId, dependency)
 
     dependency match {
-      case _ if shouldFallThroughOnEverything ||
-        rapidsConf.isMultiThreadedShuffleManagerMode => orig
       case gpuDependency: GpuShuffleDependency[K, V, C] if gpuDependency.useGPUShuffle =>
-        new GpuShuffleHandle(orig,
-          dependency.asInstanceOf[GpuShuffleDependency[K, V, V]])
+        val gpuDep = gpuDependency.asInstanceOf[GpuShuffleDependency[K, V, V]]
+        gpuDep.checksumFallback = shouldFallThroughForShuffle
+        if (rapidsConf.isMultiThreadedShuffleManagerMode) orig
+        else new GpuShuffleHandle(orig, gpuDep)
+      case _ if shouldAlwaysFallBack ||
+        rapidsConf.isMultiThreadedShuffleManagerMode => orig
       case _ => orig
     }
   }
@@ -1900,6 +1913,8 @@ class RapidsShuffleInternalManagerBase(conf: SparkConf, val isDriver: Boolean)
       context: TaskContext,
       metricsReporter: ShuffleWriteMetricsReporter): ShuffleWriter[K, V] = {
     handle match {
+      case gpu: GpuShuffleHandle[_, _] if gpu.dependency.checksumFallback =>
+        wrapped.getWriter(gpu.wrapped, mapId, context, metricsReporter)
       case gpu: GpuShuffleHandle[_, _] =>
         registerGpuShuffle(handle.shuffleId)
         new RapidsCachingWriter(
@@ -1914,6 +1929,7 @@ class RapidsShuffleInternalManagerBase(conf: SparkConf, val isDriver: Boolean)
         handle.dependency match {
           case gpuDep: GpuShuffleDependency[_, _, _]
             if gpuDep.useMultiThreadedShuffle &&
+              !gpuDep.checksumFallback &&
               rapidsConf.shuffleMultiThreadedWriterThreads > 0 =>
             // use the threaded writer if the number of threads specified is 1 or above,
             // with 0 threads we fallback to the Spark-provided writer.
@@ -1957,6 +1973,9 @@ class RapidsShuffleInternalManagerBase(conf: SparkConf, val isDriver: Boolean)
       context: TaskContext,
       metrics: ShuffleReadMetricsReporter): ShuffleReader[K, C] = {
     handle match {
+      case gpuHandle: GpuShuffleHandle[_, _] if gpuHandle.dependency.checksumFallback =>
+        ShuffleManagerShims.getReader(wrapped, gpuHandle.wrapped, startMapIndex, endMapIndex,
+          startPartition, endPartition, context, metrics)
       case gpuHandle: GpuShuffleHandle[_, _] =>
         logInfo(s"Asking map output tracker for dependency ${gpuHandle.dependency}, " +
           s"map output sizes for: ${gpuHandle.shuffleId}, parts=$startPartition-$endPartition")
@@ -1995,7 +2014,8 @@ class RapidsShuffleInternalManagerBase(conf: SparkConf, val isDriver: Boolean)
         //   would need to be made to deal with missing metrics, for example, for a regular
         //   Exchange node.
         baseHandle.dependency match {
-          case gpuDep: GpuShuffleDependency[K, C, C] if gpuDep.useMultiThreadedShuffle =>
+          case gpuDep: GpuShuffleDependency[K, C, C]
+              if gpuDep.useMultiThreadedShuffle && !gpuDep.checksumFallback =>
             // We want to use batch fetch in the non-push shuffle case. Spark
             // checks for a config to see if batch fetch is enabled (this check), and
             // it also checks when getting (potentially merged) map status from

sql-plugin/src/test/spark420/scala/com/nvidia/spark/rapids/shims/spark420/SparkShimsSuite.scala

@@ -22,7 +22,13 @@ package com.nvidia.spark.rapids.shims.spark420
 import com.nvidia.spark.rapids._
 import org.scalatest.funsuite.AnyFunSuite
 
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.internal.SQLConf
+
 class SparkShimsSuite extends AnyFunSuite with FQSuiteName {
+  private val checksumEnabledKey = RowBasedShuffleChecksumConf.ChecksumEnabledKey
+  private val fullRetryKey = RowBasedShuffleChecksumConf.ChecksumMismatchFullRetryKey
+
   test("spark shims version") {
     assert(ShimLoader.getShimVersion === SparkShimVersion(4, 2, 0))
   }
@@ -32,4 +38,30 @@ class SparkShimsSuite extends AnyFunSuite with FQSuiteName {
       classOf[com.nvidia.spark.rapids.spark420.RapidsShuffleManager].getCanonicalName)
   }
 
+  test("row-based shuffle checksum defaults to enabled") {
+    val sqlConf = new SQLConf()
+    assert(sqlConf.shuffleOrderIndependentChecksumEnabled)
+    assert(sqlConf.shuffleChecksumMismatchFullRetryEnabled)
+    assert(RowBasedShuffleChecksumConf.isEnabled(sqlConf, new SparkConf(false)))
+  }
+
+  test("row-based shuffle checksum uses SparkConf when SQLConf is unset") {
+    val sparkConf = new SparkConf(false)
+      .set(checksumEnabledKey, "false")
+      .set(fullRetryKey, "false")
+
+    assert(!RowBasedShuffleChecksumConf.isEnabled(new SQLConf(), sparkConf))
+  }
+
+  test("row-based shuffle checksum uses SQLConf when present") {
+    val sqlConf = new SQLConf()
+    sqlConf.setConfString(checksumEnabledKey, "false")
+    sqlConf.setConfString(fullRetryKey, "false")
+    val sparkConf = new SparkConf(false)
+      .set(checksumEnabledKey, "true")
+      .set(fullRetryKey, "true")
+
+    assert(!RowBasedShuffleChecksumConf.isEnabled(sqlConf, sparkConf))
+  }
+
 }