[AutoSparkUT] Fix GPU missing-file recovery guidance [reduced-it] [fast-ut] [databricks]

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

@@ -0,0 +1,44 @@
+/*
+ * 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 java.io.FileNotFoundException
+
+/**
+ * Carries the owning file path across an asynchronous reader boundary.
+ *
+ * Spark 4.x needs the path to construct `FAILED_READ_FILE.FILE_NOT_EXIST`, but a
+ * `Future.get()` otherwise exposes only the reader's `FileNotFoundException`.
+ */
+object GpuFileNotFoundException {
+  private final class WithPath(
+      val filePath: String,
+      val originalException: FileNotFoundException)
+    extends FileNotFoundException(originalException.getMessage) {
+    initCause(originalException)
+  }
+
+  def apply(filePath: String, error: FileNotFoundException): FileNotFoundException = error match {
+    case pathError: WithPath => pathError
+    case _ => new WithPath(filePath, error)
+  }
+
+  def unapply(error: FileNotFoundException): Option[(String, FileNotFoundException)] = error match {
+    case pathError: WithPath => Some((pathError.filePath, pathError.originalException))
+    case _ => None
+  }
+}

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

@@ -16,7 +16,7 @@
 
 package com.nvidia.spark.rapids
 
-import java.io.{File, IOException}
+import java.io.{File, FileNotFoundException, IOException}
 import java.net.{URI, URISyntaxException}
 import java.util.concurrent.{CompletionService, ConcurrentLinkedQueue, ExecutorCompletionService, Future, ThreadPoolExecutor, TimeUnit}
 import java.util.concurrent.atomic.{AtomicBoolean, AtomicInteger}
@@ -581,6 +581,11 @@ abstract class MultiFileCloudPartitionReaderBase(
     // An AsyncRunner wrapper used to update related metrics
     val newTaskRunner = (file: PartitionedFile) => {
       val runner = getBatchRunner(tc, file, conf, filters)
+      runner.addFailureTransformer {
+        case error: FileNotFoundException =>
+          GpuFileNotFoundException(file.filePath.toString, error)
+        case error => error
+      }
       val metrics = GpuTaskMetrics.get
       val taskId = tc.taskAttemptId()
       runner.addPreHook(() => {
@@ -1472,8 +1477,13 @@ abstract class MultiFileCoalescingPartitionReaderBase(
             // use a single buffer and slice it up for different files if we need
             val outLocal = hmb.slice(offset, fileBlockSize)
             // Third, copy the blocks for each file in parallel using background threads
-            tasks.add(threadPool.submit(
-              getBatchRunner(tc, file, outLocal, blocks, offset, batchContext)))
+            val runner = getBatchRunner(tc, file, outLocal, blocks, offset, batchContext)
+            runner.addFailureTransformer {
+              case error: FileNotFoundException =>
+                GpuFileNotFoundException(file.toString, error)
+              case error => error
+            }
+            tasks.add(threadPool.submit(runner))
             offset += fileBlockSize
           }
 

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

@@ -734,6 +734,8 @@ case class GpuOrcMultiFilePartitionReaderFactory(
           } catch {
             case e: FileNotFoundException if ignoreMissingFiles =>
               logWarning(s"Skipped missing file: ${file.filePath}", e)
+            case e: FileNotFoundException =>
+              throw GpuFileNotFoundException(file.filePath.toString, e)
           }
         }
       }

sql-plugin/src/main/scala/com/nvidia/spark/rapids/io/async/AsyncRunners.scala

@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2025, NVIDIA CORPORATION.
+ * Copyright (c) 2025-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.
@@ -22,6 +22,7 @@ import java.util.concurrent.locks.ReentrantLock
 import java.util.function.LongUnaryOperator
 
 import scala.collection.mutable
+import scala.util.control.NonFatal
 
 import com.nvidia.spark.rapids.jni.TaskPriority
 
@@ -195,6 +196,8 @@ trait AsyncRunner[T] extends Callable[AsyncResult[T]] {
     val resultData = try {
       beforeExecuteHooks.foreach { hook => hook() }
       callImpl()
+    } catch {
+      case NonFatal(error) => throw failureTransformer(error)
     } finally {
       afterExecuteHooks.foreach { hook => hook() }
     }
@@ -206,13 +209,23 @@ trait AsyncRunner[T] extends Callable[AsyncResult[T]] {
 
   private val beforeExecuteHooks = mutable.ArrayBuffer.empty[() => Unit]
   private val afterExecuteHooks = mutable.ArrayBuffer.empty[() => Unit]
+  private var failureTransformer: Throwable => Throwable = identity
 
   // Add hook to be executed right before the task execution.
   def addPreHook(hook: () => Unit): Unit = beforeExecuteHooks += hook
 
   // Add hook to be executed right after the task execution.
   def addPostHook(hook: () => Unit): Unit = afterExecuteHooks += hook
 
+  /**
+   * Adds a transformer for failures thrown by the runner body. Transformers are applied in the
+   * order they are added and are not invoked on the successful execution path.
+   */
+  def addFailureTransformer(transformer: Throwable => Throwable): Unit = {
+    val previousTransformer = failureTransformer
+    failureTransformer = error => transformer(previousTransformer(error))
+  }
+
   /**
    * This method is called when the required resource has been just acquired from pool.
    * It can be overridden by subclasses to perform actions right after the acquisition.

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

@@ -1297,7 +1297,8 @@ abstract class AbstractGpuParquetMultiFilePartitionReaderFactory(
           hasInt96Timestamps = false)
         BlockMetaWithPartFile(meta, file)
       // Throw FileNotFoundException even if `ignoreCorruptFiles` is true
-      case e: FileNotFoundException if !ignoreMissingFiles => throw e
+      case e: FileNotFoundException if !ignoreMissingFiles =>
+        throw GpuFileNotFoundException(file.filePath.toString, e)
       // If ignoreMissingFiles=true, this case will never be reached. But it's ok
       // to leave this branch here.
       case e@(_: RuntimeException | _: IOException) if ignoreCorruptFiles =>

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

@@ -45,7 +45,8 @@ abstract class GpuBatchScanExecBase(
       // return an empty RDD with 1 partition if dynamic filtering removed the only split
       sparkContext.parallelize(Array.empty[InternalRow], 1)
     } else {
-      new GpuDataSourceRDD(sparkContext, filteredPartitions, readerFactory)
+      new GpuDataSourceRDD(
+        sparkContext, filteredPartitions, readerFactory, includeRefreshHint = false)
     }
   }
 

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

@@ -16,13 +16,18 @@
 
 package com.nvidia.spark.rapids.shims
 
-import com.nvidia.spark.rapids.{FileSystemBytesReadTracker, MetricsBatchIterator, PartitionIterator}
+import java.io.FileNotFoundException
+import java.util.concurrent.ExecutionException
+
+import com.nvidia.spark.rapids.{FileSystemBytesReadTracker, GpuFileNotFoundException,
+  MetricsBatchIterator, PartitionIterator}
 import com.nvidia.spark.rapids.ScalableTaskCompletion.onTaskCompletion
 
 import org.apache.spark.{InterruptibleIterator, Partition, SparkContext, SparkException, TaskContext}
 import org.apache.spark.rdd.RDD
 import org.apache.spark.sql.catalyst.InternalRow
 import org.apache.spark.sql.connector.read.{InputPartition, PartitionReaderFactory}
+import org.apache.spark.sql.execution.datasources.FilePartition
 import org.apache.spark.sql.vectorized.ColumnarBatch
 
 /**
@@ -33,7 +38,8 @@ import org.apache.spark.sql.vectorized.ColumnarBatch
 class GpuDataSourceRDD(
     sc: SparkContext,
     @transient private val inputPartitions: Seq[Seq[InputPartition]],
-    partitionReaderFactory: PartitionReaderFactory
+    partitionReaderFactory: PartitionReaderFactory,
+    includeRefreshHint: Boolean
 ) extends RDD[InternalRow](sc, Nil) {
   import GpuDataSourceRDD.GpuDataSourceRDDPartition
 
@@ -59,13 +65,25 @@ class GpuDataSourceRDD(
       private val inputPartitions = castPartition(split).inputPartitions
       private var currentIter: Option[Iterator[Object]] = None
       private var currentIndex: Int = 0
+      private var currentInputPartition: InputPartition = _
 
-      override def hasNext: Boolean = {
+      override def hasNext: Boolean = try {
         val result = currentIter.exists(_.hasNext) || advanceToNextIter()
         if (!result) {
           bytesReadTracker.update()
         }
         result
+      } catch {
+        case e: FileNotFoundException =>
+          throw GpuDataSourceRDD.missingFileError(
+            e, includeRefreshHint, currentInputPartition)
+        case e: ExecutionException =>
+          e.getCause match {
+            case cause: FileNotFoundException =>
+              throw GpuDataSourceRDD.missingFileError(
+                cause, includeRefreshHint, currentInputPartition)
+            case _ => throw e
+          }
       }
 
       override def next(): Object = {
@@ -74,6 +92,17 @@ class GpuDataSourceRDD(
             throw new NoSuchElementException("No more elements")
           }
           currentIter.get.next()
+        } catch {
+          case e: FileNotFoundException =>
+            throw GpuDataSourceRDD.missingFileError(
+              e, includeRefreshHint, currentInputPartition)
+          case e: ExecutionException =>
+            e.getCause match {
+              case cause: FileNotFoundException =>
+                throw GpuDataSourceRDD.missingFileError(
+                  cause, includeRefreshHint, currentInputPartition)
+              case _ => throw e
+            }
         } finally {
           bytesReadTracker.update()
         }
@@ -84,6 +113,7 @@ class GpuDataSourceRDD(
           false
         } else {
           val inputPartition = inputPartitions(currentIndex)
+          currentInputPartition = inputPartition
           currentIndex += 1
 
           // TODO: SPARK-25083 remove the type erasure hack in data source scan
@@ -112,14 +142,35 @@ class GpuDataSourceRDD(
 }
 
 object GpuDataSourceRDD {
+  private def missingFileError(
+      error: FileNotFoundException,
+      includeRefreshHint: Boolean,
+      inputPartition: InputPartition): Throwable = {
+    val (filePath, originalError) = error match {
+      case GpuFileNotFoundException(path, originalException) =>
+        (Some(path), originalException)
+      case _ =>
+        (singleFilePath(inputPartition), error)
+    }
+    MissingFileErrorShim.convert(filePath, originalError, includeRefreshHint)
+  }
+
+  private def singleFilePath(inputPartition: InputPartition): Option[String] = {
+    Option(inputPartition).collect { case filePartition: FilePartition =>
+      SparkShimImpl.getPartitionFiles(filePartition)
+    }.filter(_.length == 1).map(_.head.filePath.toString)
+  }
+
   private case class GpuDataSourceRDDPartition(
       override val index: Int,
       inputPartitions: Seq[InputPartition]) extends Partition
 
   def apply(
       sc: SparkContext,
       inputPartitions: Seq[InputPartition],
-      partitionReaderFactory: PartitionReaderFactory): GpuDataSourceRDD = {
-    new GpuDataSourceRDD(sc, inputPartitions.map(Seq(_)), partitionReaderFactory)
+      partitionReaderFactory: PartitionReaderFactory,
+      includeRefreshHint: Boolean): GpuDataSourceRDD = {
+    new GpuDataSourceRDD(
+      sc, inputPartitions.map(Seq(_)), partitionReaderFactory, includeRefreshHint)
   }
 }

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

@@ -281,7 +281,8 @@ case class GpuAvroMultiFilePartitionReaderFactory(
               logWarning(s"Skipped missing file: ${file.filePath}", e)
               AvroBlockMeta(null, 0L, Seq.empty)
             // Throw FileNotFoundException even if `ignoreCorruptFiles` is true
-            case e: FileNotFoundException if !ignoreMissingFiles => throw e
+            case e: FileNotFoundException if !ignoreMissingFiles =>
+              throw GpuFileNotFoundException(file.filePath.toString, e)
             case e@(_: RuntimeException | _: IOException) if ignoreCorruptFiles =>
               logWarning(
                 s"Skipped the rest of the content in the corrupted file: ${file.filePath}", e)

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

@@ -612,7 +612,11 @@ case class GpuFileSourceScanExec(
       logDebug(s"Using Datasource RDD, files are: " +
         s"${prunedPartitions.flatMap(FilePartitionShims.getFiles).mkString(",")}")
       // note we use the v2 DataSourceRDD instead of FileScanRDD so we don't have to copy more code
-      GpuDataSourceRDD(relation.sparkSession.sparkContext, locatedPartitions, readerFactory)
+      GpuDataSourceRDD(
+        relation.sparkSession.sparkContext,
+        locatedPartitions,
+        readerFactory,
+        includeRefreshHint = true)
     }
   }
 

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

@@ -0,0 +1,79 @@
+/*
+ * 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.
+ */
+
+/*** spark-rapids-shim-json-lines
+{"spark": "330"}
+{"spark": "330db"}
+{"spark": "331"}
+{"spark": "332"}
+{"spark": "332db"}
+{"spark": "333"}
+{"spark": "334"}
+{"spark": "340"}
+{"spark": "341"}
+{"spark": "342"}
+{"spark": "343"}
+{"spark": "344"}
+{"spark": "350"}
+{"spark": "350db143"}
+{"spark": "351"}
+{"spark": "352"}
+{"spark": "353"}
+{"spark": "354"}
+{"spark": "355"}
+{"spark": "356"}
+{"spark": "357"}
+{"spark": "358"}
+{"spark": "359"}
+spark-rapids-shim-json-lines ***/
+package com.nvidia.spark.rapids.shims
+
+import java.io.FileNotFoundException
+
+import org.apache.spark.sql.connector.read.PartitionReaderFactory
+
+object MissingFileErrorShim {
+  private val RECREATE_HINT = "recreating the Dataset/DataFrame involved"
+  private val REFRESH_HINT = "REFRESH TABLE"
+
+  def wrapReaderFactory(readerFactory: PartitionReaderFactory): PartitionReaderFactory =
+    readerFactory
+
+  def convert(
+      filePath: Option[String],
+      error: FileNotFoundException,
+      includeRefreshHint: Boolean): Throwable = {
+    val message = Option(error.getMessage).getOrElse(error.toString)
+    if (message.contains(RECREATE_HINT) &&
+        (!includeRefreshHint || message.contains(REFRESH_HINT))) {
+      error
+    } else {
+      val recoveryHint = if (includeRefreshHint) {
+        "It is possible the underlying files have been updated. " +
+          "You can explicitly invalidate the cache in Spark by " +
+          "running 'REFRESH TABLE tableName' command in SQL or " +
+          "by recreating the Dataset/DataFrame involved."
+      } else {
+        "It is possible the underlying files have been updated. " +
+          "You can explicitly invalidate the cache in Spark by " +
+          "recreating the Dataset/DataFrame involved."
+      }
+      val enrichedException = new FileNotFoundException(s"$message\n$recoveryHint")
+      enrichedException.initCause(error)
+      enrichedException
+    }
+  }
+}

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

@@ -247,7 +247,8 @@ case class GpuBatchScanExec(
       }
 
       // Use the finalized partitions so padded and replicated inputs match outputPartitioning.
-      new GpuDataSourceRDD(sparkContext, finalPartitions, readerFactory)
+      new GpuDataSourceRDD(
+        sparkContext, finalPartitions, readerFactory, includeRefreshHint = false)
     }
     postDriverMetrics()
     rdd