Merge branch 'main' into fix/parallel-unit-tests

Author: res-life

delta-lake/common/src/main/delta-33x-41x/scala/com/nvidia/spark/rapids/delta/common/GpuDeltaParquetFileFormatBase2.scala

@@ -312,7 +312,8 @@ class GpuDeltaParquetFileFormatBase2(
             val (rowGroupOffsets, rowGroupNumRows) =
               RapidsDeletionVectors.getRowGroupMetadata(chunkedBlocks)
             val maybeDvInfo = maybeSerializedDV.map(serializedDV =>
-              new DeletionVector.DeletionVectorInfo(serializedDV, rowGroupOffsets, rowGroupNumRows))
+              new DeletionVector.DeletionVectorInfo(serializedDV,
+                false, rowGroupOffsets, rowGroupNumRows))
 
             val hostBuf = dataBuffer.getDataHostBuffer()
             // Duplicate request is ok, and start to use the GPU just after the host
@@ -686,6 +687,7 @@ class GpuDeltaParquetFileFormatBase2(
               .map(spillableDvInfo =>
                 new DeletionVector.DeletionVectorInfo(
                   spillableDvInfo.serializedBitmap.getDataHostBuffer(),
+                  false,
                   spillableDvInfo.rowGroupOffsets,
                   spillableDvInfo.rowGroupNumRows
                 ))
@@ -1193,7 +1195,8 @@ class GpuDeltaParquetFileFormatBase2(
           .zip(batchExtra.perFileEntries)
           .map { case (loaded, entry) =>
             new DeletionVector.DeletionVectorInfo(
-              loaded.gpuBitmap.getDataHostBuffer(), entry.rowGroupOffsets, entry.rowGroupNumRows)
+              loaded.gpuBitmap.getDataHostBuffer(),
+              false, entry.rowGroupOffsets, entry.rowGroupNumRows)
           }.toArray
         // MakeParquetTableWithDVProducer closes the dataBuffer and the bitmaps in dvInfos.
         MakeParquetTableWithDVProducer(useChunkedReader, maxChunkedReaderMemoryUsageSizeBytes,

delta-lake/delta-spark400db173/src/main/scala/com/nvidia/spark/rapids/delta/GpuDeltaParquetFileFormatNativeDV.scala

@@ -321,7 +321,8 @@ case class GpuDeltaParquetFileFormatNativeDV(
             val (rowGroupOffsets, rowGroupNumRows) =
               RapidsDeletionVectors.getRowGroupMetadata(chunkedBlocks)
             val maybeDvInfo = maybeSerializedDV.map(serializedDV =>
-              new DeletionVector.DeletionVectorInfo(serializedDV, rowGroupOffsets, rowGroupNumRows))
+              new DeletionVector.DeletionVectorInfo(serializedDV,
+                false, rowGroupOffsets, rowGroupNumRows))
 
             val hostBuf = dataBuffer.getDataHostBuffer()
             // Duplicate request is ok, and start to use the GPU just after the host
@@ -752,6 +753,7 @@ case class GpuDeltaParquetFileFormatNativeDV(
               .map(spillableDvInfo =>
                 new DeletionVector.DeletionVectorInfo(
                   spillableDvInfo.serializedBitmap.getDataHostBuffer(),
+                  false,
                   spillableDvInfo.rowGroupOffsets,
                   spillableDvInfo.rowGroupNumRows
                 ))
@@ -1350,7 +1352,8 @@ case class GpuDeltaParquetFileFormatNativeDV(
           .zip(batchExtra.perFileEntries)
           .map { case (loaded, entry) =>
             new DeletionVector.DeletionVectorInfo(
-              loaded.gpuBitmap.getDataHostBuffer(), entry.rowGroupOffsets, entry.rowGroupNumRows)
+              loaded.gpuBitmap.getDataHostBuffer(),
+              false, entry.rowGroupOffsets, entry.rowGroupNumRows)
           }.toArray
         // MakeParquetTableWithDVProducer closes the dataBuffer and the bitmaps in dvInfos.
         MakeParquetTableWithDVProducer(useChunkedReader, maxChunkedReaderMemoryUsageSizeBytes,

docs/additional-functionality/advanced_configs.md

@@ -50,7 +50,7 @@ Name | Description | Default Value | Applicable at
 <a name="memory.gpu.reserve"></a>spark.rapids.memory.gpu.reserve|The amount of GPU memory that should remain unallocated by RMM and left for system use such as memory needed for kernels and kernel launches.|671088640|Startup
 <a name="memory.gpu.state.debug"></a>spark.rapids.memory.gpu.state.debug|To better recover from out of memory errors, RMM will track several states for the threads that interact with the GPU. This provides a log of those state transitions to aid in debugging it. STDOUT or STDERR will have the logging go there empty string will disable logging and anything else will be treated as a file to write the logs to.||Startup
 <a name="memory.gpu.unspill.enabled"></a>spark.rapids.memory.gpu.unspill.enabled|When a spilled GPU buffer is needed again, should it be unspilled, or only copied back into GPU memory temporarily. Unspilling may be useful for GPU buffers that are needed frequently, for example, broadcast variables; however, it may also increase GPU memory usage|false|Startup
-<a name="perfio.gcs.enabled"></a>spark.rapids.perfio.gcs.enabled|Controls the Google Cloud Storage reader for improved performance in certain queries. When true, enables it and throws at startup if google-cloud-storage classes are not on the classpath. When false, disables it unconditionally. When unset (default), enables it opportunistically if google-cloud-storage classes are found, otherwise falls back to the configured GCS connector with a warning. The presence of com.google.cloud:google-cloud-storage on the executor classpath is required.|None|Startup
+<a name="perfio.gcs.enabled"></a>spark.rapids.perfio.gcs.enabled|Controls the Google Cloud Storage reader for improved performance in certain queries. When true, enables it. When false, disables it unconditionally. When unset (default), enables it on Google Cloud Dataproc when compatible connector classes are available; outside Dataproc it remains disabled. Explicit enablement fails at startup if the required classes are unavailable.|None|Startup
 <a name="perfio.s3.enabled"></a>spark.rapids.perfio.s3.enabled|Controls the AWS S3 reader for improved performance in certain queries. When true, enables it and throws at startup if no compatible HTTP client is on the classpath. When false, disables it unconditionally. When unset (default), enables it opportunistically if a compatible HTTP client is found, otherwise falls back to S3A with a warning. The presence of AWS SDK packages for Netty and/or CRT HTTP clients on the classpath is required. You can use Spark submit option `--packages software.amazon.awssdk:s3:2.22.12,software.amazon.awssdk:aws-crt-client:2.22.12` to achieve this. See https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/crt-based-s3-client.html#crt-based-s3-client-depend|None|Startup
 <a name="python.concurrentPythonWorkers"></a>spark.rapids.python.concurrentPythonWorkers|Set the number of Python worker processes that can execute concurrently per GPU. Python worker processes may temporarily block when the number of concurrent Python worker processes started by the same executor exceeds this amount. Allowing too many concurrent tasks on the same GPU may lead to GPU out of memory errors. >0 means enabled, while <=0 means unlimited|0|Runtime
 <a name="python.memory.gpu.allocFraction"></a>spark.rapids.python.memory.gpu.allocFraction|The fraction of total GPU memory that should be initially allocated for pooled memory for all the Python workers. It supposes to be less than (1 - $(spark.rapids.memory.gpu.allocFraction)), since the executor will share the GPU with its owning Python workers. Half of the rest will be used if not specified|None|Runtime

jenkins/databricks/init_cudf_udf.sh

@@ -20,7 +20,6 @@
 
 set -ex
 
-# TODO: https://github.com/NVIDIA/cudf-spark/issues/15386
 CUDF_VER=${CUDF_VER:-26.10.0-SNAPSHOT}
 CUDA_VER=${CUDA_VER:-12.9}
 

pom.xml

@@ -987,10 +987,9 @@
         <spark.version.classifier>spark${buildver}</spark.version.classifier>
         <cuda.version>cuda12</cuda.version>
         <jni.classifier>${cuda.version}</jni.classifier>
-        <!-- TODO: https://github.com/NVIDIA/cudf-spark/issues/15386 -->
-        <spark-rapids-jni.version>26.08.0-SNAPSHOT</spark-rapids-jni.version>
-        <spark-rapids-private.version>26.08.0-SNAPSHOT</spark-rapids-private.version>
-        <spark-rapids-hybrid.version>26.08.0-SNAPSHOT</spark-rapids-hybrid.version>
+        <spark-rapids-jni.version>26.10.0-SNAPSHOT</spark-rapids-jni.version>
+        <spark-rapids-private.version>26.10.0-SNAPSHOT</spark-rapids-private.version>
+        <spark-rapids-hybrid.version>26.10.0-SNAPSHOT</spark-rapids-hybrid.version>
         <scala.binary.version>2.12</scala.binary.version>
         <scala.recompileMode>incremental</scala.recompileMode>
         <scala.version>2.12.21</scala.version>

scala2.13/pom.xml

@@ -987,10 +987,9 @@
         <spark.version.classifier>spark${buildver}</spark.version.classifier>
         <cuda.version>cuda12</cuda.version>
         <jni.classifier>${cuda.version}</jni.classifier>
-        <!-- TODO: https://github.com/NVIDIA/cudf-spark/issues/15386 -->
-        <spark-rapids-jni.version>26.08.0-SNAPSHOT</spark-rapids-jni.version>
-        <spark-rapids-private.version>26.08.0-SNAPSHOT</spark-rapids-private.version>
-        <spark-rapids-hybrid.version>26.08.0-SNAPSHOT</spark-rapids-hybrid.version>
+        <spark-rapids-jni.version>26.10.0-SNAPSHOT</spark-rapids-jni.version>
+        <spark-rapids-private.version>26.10.0-SNAPSHOT</spark-rapids-private.version>
+        <spark-rapids-hybrid.version>26.10.0-SNAPSHOT</spark-rapids-hybrid.version>
         <scala.binary.version>2.13</scala.binary.version>
         <scala.recompileMode>incremental</scala.recompileMode>
         <scala.version>2.13.18</scala.version>

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

@@ -19,32 +19,46 @@ package com.nvidia.spark.rapids.shims
 import java.nio.ByteBuffer
 
 import scala.collection.mutable.ArrayBuffer
+import scala.util.control.NonFatal
 
 import com.nvidia.spark.rapids.Arm.withResource
 import com.nvidia.spark.rapids.OrcOutputStripe
-import com.nvidia.spark.rapids.RapidsPluginImplicits._
 import org.apache.hadoop.conf.Configuration
 import org.apache.orc.{CompressionCodec, CompressionKind, DataReader, OrcConf, OrcFile, OrcProto, PhysicalWriter, Reader, StripeInformation, TypeDescription}
 import org.apache.orc.impl.{BufferChunk, DataReaderProperties, InStream, OrcCodecPool, OutStream, ReaderImpl, SchemaEvolution}
 import org.apache.orc.impl.RecordReaderImpl.SargApplier
 import org.apache.orc.impl.reader.StripePlanner
 import org.apache.orc.impl.writer.StreamOptions
 
-trait OrcShims320untilAllBase {
+import org.apache.spark.internal.Logging
+
+trait OrcShims320untilAllBase extends Logging {
 
   // the ORC Reader in non-CDH Spark is closeable
   def withReader[T <: Reader, V](r: T)(block: T => V): V = {
     try {
       block(r)
     } finally {
-      r.safeClose()
+      closeReader(r)
     }
   }
 
   // the ORC Reader in non-CDH Spark is closeable
   def closeReader(reader: Reader): Unit = {
-    if(reader != null) {
-      reader.close()
+    if (reader != null) {
+      // Close without being aborted by a pending interrupt from task cancellation, then restore
+      // the interrupt status for the caller. This matches Spark's SPARK-57958 ORC reader hardening.
+      val interrupted = Thread.interrupted()
+      try {
+        reader.close()
+      } catch {
+        case NonFatal(t) =>
+          logWarning("Failed to close the ORC reader", t)
+      } finally {
+        if (interrupted) {
+          Thread.currentThread().interrupt()
+        }
+      }
     }
   }
 

tests/src/test/scala/com/nvidia/spark/rapids/shims/OrcShimsSuite.scala

@@ -0,0 +1,64 @@
+/*
+ * 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.shims
+
+import java.io.IOException
+
+import org.apache.orc.Reader
+import org.mockito.Mockito.{doAnswer, doThrow, verify}
+import org.scalatest.funsuite.AnyFunSuite
+import org.scalatestplus.mockito.MockitoSugar
+
+class OrcShimsSuite extends AnyFunSuite with MockitoSugar {
+
+  test("closeReader clears and restores interrupt status") {
+    val reader = mock[Reader]
+    var closedWithInterruptSet = true
+    doAnswer { _ =>
+      closedWithInterruptSet = Thread.currentThread().isInterrupted
+      null
+    }.when(reader).close()
+
+    Thread.interrupted()
+    Thread.currentThread().interrupt()
+    try {
+      OrcShims.closeReader(reader)
+
+      assert(!closedWithInterruptSet)
+      assert(Thread.currentThread().isInterrupted)
+      verify(reader).close()
+    } finally {
+      Thread.interrupted()
+    }
+  }
+
+  test("closeReader restores interrupt status after non-fatal close failure") {
+    val reader = mock[Reader]
+    doThrow(new IOException("close failed")).when(reader).close()
+
+    Thread.interrupted()
+    Thread.currentThread().interrupt()
+    try {
+      OrcShims.closeReader(reader)
+
+      assert(Thread.currentThread().isInterrupted)
+      verify(reader).close()
+    } finally {
+      Thread.interrupted()
+    }
+  }
+}