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

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

@@ -16,6 +16,9 @@
 
 package com.nvidia.spark.rapids.shims
 
+import java.io.FileNotFoundException
+import java.util.concurrent.ExecutionException
+
 import com.nvidia.spark.rapids.{FileSystemBytesReadTracker, MetricsBatchIterator, PartitionIterator}
 import com.nvidia.spark.rapids.ScalableTaskCompletion.onTaskCompletion
 
@@ -33,7 +36,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 = false
 ) extends RDD[InternalRow](sc, Nil) {
   import GpuDataSourceRDD.GpuDataSourceRDDPartition
 
@@ -60,12 +64,21 @@ class GpuDataSourceRDD(
       private var currentIter: Option[Iterator[Object]] = None
       private var currentIndex: Int = 0
 
-      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.withRecoveryHint(e, includeRefreshHint)
+        case e: ExecutionException =>
+          e.getCause match {
+            case cause: FileNotFoundException =>
+              throw GpuDataSourceRDD.withRecoveryHint(cause, includeRefreshHint)
+            case _ => throw e
+          }
       }
 
       override def next(): Object = {
@@ -74,6 +87,15 @@ class GpuDataSourceRDD(
             throw new NoSuchElementException("No more elements")
           }
           currentIter.get.next()
+        } catch {
+          case e: FileNotFoundException =>
+            throw GpuDataSourceRDD.withRecoveryHint(e, includeRefreshHint)
+          case e: ExecutionException =>
+            e.getCause match {
+              case cause: FileNotFoundException =>
+                throw GpuDataSourceRDD.withRecoveryHint(cause, includeRefreshHint)
+              case _ => throw e
+            }
         } finally {
           bytesReadTracker.update()
         }
@@ -112,14 +134,43 @@ class GpuDataSourceRDD(
 }
 
 object GpuDataSourceRDD {
+  private val RECREATE_HINT = "recreating the Dataset/DataFrame involved"
+  private val REFRESH_HINT = "REFRESH TABLE"
+
+  private def withRecoveryHint(
+      e: FileNotFoundException,
+      includeRefreshHint: Boolean): FileNotFoundException = {
+    val message = Option(e.getMessage).getOrElse(e.toString)
+    if (message.contains(RECREATE_HINT) &&
+        (!includeRefreshHint || message.contains(REFRESH_HINT))) {
+      e
+    } 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(e)
+      enrichedException
+    }
+  }
+
   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 = false): GpuDataSourceRDD = {
+    new GpuDataSourceRDD(
+      sc, inputPartitions.map(Seq(_)), partitionReaderFactory, includeRefreshHint)
   }
 }

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)
     }
   }
 

tests/src/test/scala/com/nvidia/spark/rapids/FileSystemBytesReadTrackerSuite.scala

@@ -16,6 +16,8 @@
 
 package com.nvidia.spark.rapids
 
+import java.io.FileNotFoundException
+import java.util.concurrent.ExecutionException
 import java.util.concurrent.atomic.AtomicLong
 
 import com.nvidia.spark.rapids.shims.GpuDataSourceRDD
@@ -127,6 +129,56 @@ class FileSystemBytesReadTrackerSuite extends AnyFunSuite with MockitoSugar {
     }
   }
 
+  Seq(
+    ("direct V2", false, () => new FileNotFoundException("missing ORC file")),
+    ("wrapped V1", true,
+      () => new ExecutionException(new FileNotFoundException("missing ORC file")))
+  ).foreach { case (name, includeRefreshHint, failure) =>
+    test(s"GPU datasource RDD enriches next() missing-file failures - $name") {
+      withTaskContext { context =>
+        val inputPartition = new InputPartition {}
+        val factory = new PartitionReaderFactory {
+          override def createReader(partition: InputPartition) =
+            throw new UnsupportedOperationException
+
+          override def createColumnarReader(partition: InputPartition) =
+            new PartitionReader[ColumnarBatch] {
+              private var hasNext = true
+
+              override def next(): Boolean = {
+                if (hasNext) {
+                  hasNext = false
+                  true
+                } else {
+                  false
+                }
+              }
+
+              override def get(): ColumnarBatch = {
+                statistics.incrementBytesRead(7L)
+                throw failure()
+              }
+
+              override def close(): Unit = {}
+            }
+
+          override def supportColumnarReads(partition: InputPartition): Boolean = true
+        }
+        val rdd = GpuDataSourceRDD(
+          mock[SparkContext], Seq(inputPartition), factory, includeRefreshHint)
+        val iterator = rdd.compute(rdd.partitions.head, context)
+
+        assert(iterator.hasNext)
+        val error = intercept[FileNotFoundException](iterator.next())
+        assert(error.getMessage.contains("recreating the Dataset/DataFrame involved"))
+        assert(error.getMessage.contains("REFRESH TABLE") === includeRefreshHint)
+        assert(error.getCause.isInstanceOf[FileNotFoundException])
+        assert(context.taskMetrics().inputMetrics.bytesRead == 7L)
+        context.markTaskComplete()
+      }
+    }
+  }
+
   test("GPU datasource RDD flushes bytes when a task stops before consuming a batch") {
     withTaskContext { context =>
       val inputPartition = new InputPartition {}

tests/src/test/spark330/scala/org/apache/spark/sql/rapids/suites/RapidsMetadataCacheSuite.scala

@@ -19,9 +19,54 @@
 spark-rapids-shim-json-lines ***/
 package org.apache.spark.sql.rapids.suites
 
-import org.apache.spark.sql.{MetadataCacheV1Suite, MetadataCacheV2Suite}
+import com.nvidia.spark.rapids.{RapidsConf, RapidsReaderType}
+
+import org.apache.spark.SparkException
+import org.apache.spark.sql.{MetadataCacheSuite, MetadataCacheV1Suite, MetadataCacheV2Suite}
 import org.apache.spark.sql.rapids.utils.RapidsSQLTestsTrait
 
+private[suites] trait RapidsMetadataCacheRecoveryHintTests {
+  self: MetadataCacheSuite with RapidsSQLTestsTrait =>
+
+  protected def expectRefreshHint: Boolean
+
+  private def exceptionMessages(error: Throwable): String = {
+    val messages = new StringBuilder
+    var current = error
+    while (current != null) {
+      messages.append(current.toString)
+      messages.append('\n')
+      current = current.getCause
+    }
+    messages.toString()
+  }
+
+  Seq(RapidsReaderType.COALESCING, RapidsReaderType.MULTITHREADED).foreach { readerType =>
+    testRapids(s"missing ORC file includes recovery guidance - $readerType") {
+      withSQLConf(RapidsConf.ORC_READER_TYPE.key -> readerType.toString) {
+        withTempPath { location =>
+          spark.range(start = 0, end = 100, step = 1, numPartitions = 3)
+            .write.orc(location.getAbsolutePath)
+
+          val df = spark.read.orc(location.getAbsolutePath)
+          assert(df.count() == 100)
+          deleteOneFileInDirectory(location)
+
+          val messages = exceptionMessages(intercept[SparkException](df.count()))
+          assert(messages.contains("recreating the Dataset/DataFrame involved"))
+          assert(messages.contains("REFRESH TABLE") === expectRefreshHint)
+        }
+      }
+    }
+  }
+}
+
 class RapidsMetadataCacheV1Suite extends MetadataCacheV1Suite with RapidsSQLTestsTrait
+    with RapidsMetadataCacheRecoveryHintTests {
+  override protected val expectRefreshHint: Boolean = true
+}
 
 class RapidsMetadataCacheV2Suite extends MetadataCacheV2Suite with RapidsSQLTestsTrait
+    with RapidsMetadataCacheRecoveryHintTests {
+  override protected val expectRefreshHint: Boolean = false
+}

tests/src/test/spark330/scala/org/apache/spark/sql/rapids/utils/RapidsTestSettings.scala

@@ -372,19 +372,7 @@ class RapidsTestSettings extends BackendTestSettings {
   enableSuite[RapidsFileSourceSQLInsertTestSuite]
   enableSuite[RapidsDSV2SQLInsertTestSuite]
   enableSuite[RapidsMetadataCacheV1Suite]
-    .exclude("SPARK-16336,SPARK-27961 Suggest fixing FileNotFoundException",
-      KNOWN_ISSUE("https://github.com/NVIDIA/cudf-spark/issues/15511. " +
-        "Recovery trigger: GPU ORC V1 missing-file errors include Spark-equivalent recreate " +
-        "guidance; P1."))
-    .exclude("SPARK-16337 temporary view refresh",
-      KNOWN_ISSUE("https://github.com/NVIDIA/cudf-spark/issues/15511. " +
-        "Recovery trigger: GPU ORC V1 missing-file errors include Spark-equivalent REFRESH " +
-        "and recreate guidance; P1."))
   enableSuite[RapidsMetadataCacheV2Suite]
-    .exclude("SPARK-16336,SPARK-27961 Suggest fixing FileNotFoundException",
-      KNOWN_ISSUE("https://github.com/NVIDIA/cudf-spark/issues/15511. " +
-        "Recovery trigger: GPU ORC V2 missing-file errors include Spark-equivalent recreate " +
-        "guidance; P1."))
   enableSuite[RapidsFileSourceStrategySuite]
     .exclude("partitioned table - after scan filters", ADJUST_UT("Replaced by testRapids version that checks GpuFilterExec residual filters."))
     .exclude("[SPARK-16818] partition pruned file scans implement sameResult correctly", KNOWN_ISSUE("https://github.com/NVIDIA/cudf-spark/issues/15161"))