diff --git a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuFileNotFoundException.scala b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuFileNotFoundException.scala new file mode 100644 index 00000000000..c7aef328fe6 --- /dev/null +++ b/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 + } +} diff --git a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuMultiFileReader.scala b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuMultiFileReader.scala index 892e151594e..82706bec398 100644 --- a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuMultiFileReader.scala +++ b/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 } diff --git a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuOrcScan.scala b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuOrcScan.scala index c23c5868863..515b4b5242e 100644 --- a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/GpuOrcScan.scala +++ b/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) } } } diff --git a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/io/async/AsyncRunners.scala b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/io/async/AsyncRunners.scala index 29f8a4debcf..b0e7f3f0239 100644 --- a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/io/async/AsyncRunners.scala +++ b/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,6 +209,7 @@ 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 @@ -213,6 +217,15 @@ trait AsyncRunner[T] extends Callable[AsyncResult[T]] { // 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. diff --git a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/parquet/GpuParquetScan.scala b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/parquet/GpuParquetScan.scala index 58eab271635..42ec45e682a 100644 --- a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/parquet/GpuParquetScan.scala +++ b/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 => diff --git a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/shims/GpuBatchScanExecBase.scala b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/shims/GpuBatchScanExecBase.scala index 81c11453b61..020a61a7558 100644 --- a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/shims/GpuBatchScanExecBase.scala +++ b/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) } } diff --git a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/shims/GpuDataSourceRDD.scala b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/shims/GpuDataSourceRDD.scala index b52266099f1..d70dd24c775 100644 --- a/sql-plugin/src/main/scala/com/nvidia/spark/rapids/shims/GpuDataSourceRDD.scala +++ b/sql-plugin/src/main/scala/com/nvidia/spark/rapids/shims/GpuDataSourceRDD.scala @@ -16,16 +16,18 @@ package com.nvidia.spark.rapids.shims -import java.util.concurrent.ConcurrentHashMap +import java.io.FileNotFoundException +import java.util.concurrent.{ConcurrentHashMap, ExecutionException} +import com.nvidia.spark.rapids.{FileSystemBytesReadTracker, GpuFileNotFoundException} import com.nvidia.spark.rapids.Arm.closeOnExcept -import com.nvidia.spark.rapids.FileSystemBytesReadTracker 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, PartitionReader, PartitionReaderFactory} +import org.apache.spark.sql.execution.datasources.FilePartition import org.apache.spark.sql.rapids.execution.TrampolineUtil import org.apache.spark.sql.vectorized.ColumnarBatch @@ -63,6 +65,7 @@ class GpuDataSourceRDD( sc: SparkContext, @transient private val inputPartitions: Seq[Seq[InputPartition]], partitionReaderFactory: PartitionReaderFactory, + includeRefreshHint: Boolean, customMetricsFactory: GpuDataSourceCustomMetricsFactory = NoopGpuDataSourceCustomMetricsFactory ) extends RDD[InternalRow](sc, Nil) { @@ -94,20 +97,45 @@ 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 = { - if (!hasNext) { - throw new NoSuchElementException("No more elements") + try { + if (!hasNext) { + 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 + } } - currentIter.get.next() } private def advanceToNextIter(): Boolean = { @@ -115,6 +143,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 @@ -236,6 +265,25 @@ 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 @@ -243,7 +291,9 @@ object GpuDataSourceRDD { 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) } } diff --git a/sql-plugin/src/main/scala/org/apache/spark/sql/rapids/GpuAvroScan.scala b/sql-plugin/src/main/scala/org/apache/spark/sql/rapids/GpuAvroScan.scala index a77d2d12f14..6f558542128 100644 --- a/sql-plugin/src/main/scala/org/apache/spark/sql/rapids/GpuAvroScan.scala +++ b/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) diff --git a/sql-plugin/src/main/scala/org/apache/spark/sql/rapids/GpuFileSourceScanExec.scala b/sql-plugin/src/main/scala/org/apache/spark/sql/rapids/GpuFileSourceScanExec.scala index 2e16a1ec8b9..2284d3a123b 100644 --- a/sql-plugin/src/main/scala/org/apache/spark/sql/rapids/GpuFileSourceScanExec.scala +++ b/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) } } diff --git a/sql-plugin/src/main/spark330/scala/com/nvidia/spark/rapids/shims/MissingFileErrorShim.scala b/sql-plugin/src/main/spark330/scala/com/nvidia/spark/rapids/shims/MissingFileErrorShim.scala new file mode 100644 index 00000000000..58703665762 --- /dev/null +++ b/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 + } + } +} diff --git a/sql-plugin/src/main/spark340/scala/com/nvidia/spark/rapids/shims/GpuBatchScanExec.scala b/sql-plugin/src/main/spark340/scala/com/nvidia/spark/rapids/shims/GpuBatchScanExec.scala index bd64876f913..418d760ff20 100644 --- a/sql-plugin/src/main/spark340/scala/com/nvidia/spark/rapids/shims/GpuBatchScanExec.scala +++ b/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 diff --git a/sql-plugin/src/main/spark350db143/scala/com/nvidia/spark/rapids/shims/GpuBatchScanExec.scala b/sql-plugin/src/main/spark350db143/scala/com/nvidia/spark/rapids/shims/GpuBatchScanExec.scala index 0490d0108d7..48db44d64ed 100644 --- a/sql-plugin/src/main/spark350db143/scala/com/nvidia/spark/rapids/shims/GpuBatchScanExec.scala +++ b/sql-plugin/src/main/spark350db143/scala/com/nvidia/spark/rapids/shims/GpuBatchScanExec.scala @@ -144,7 +144,8 @@ case class GpuBatchScanExec( } } - override lazy val readerFactory: PartitionReaderFactory = batch.createReaderFactory() + override lazy val readerFactory: PartitionReaderFactory = + MissingFileErrorShim.wrapReaderFactory(batch.createReaderFactory()) override lazy val inputRDD: RDD[InternalRow] = { scan.metrics = allMetrics @@ -254,7 +255,8 @@ case class GpuBatchScanExec( sparkContext, finalPartitions, readerFactory, - new Spark4GpuDataSourceCustomMetricsFactory(scanCustomSQLMetrics)) + includeRefreshHint = false, + customMetricsFactory = new Spark4GpuDataSourceCustomMetricsFactory(scanCustomSQLMetrics)) } postDriverMetrics() rdd diff --git a/sql-plugin/src/main/spark400/scala/com/nvidia/spark/rapids/shims/MissingFileErrorShim.scala b/sql-plugin/src/main/spark400/scala/com/nvidia/spark/rapids/shims/MissingFileErrorShim.scala new file mode 100644 index 00000000000..47ea1725a52 --- /dev/null +++ b/sql-plugin/src/main/spark400/scala/com/nvidia/spark/rapids/shims/MissingFileErrorShim.scala @@ -0,0 +1,109 @@ +/* + * 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": "400"} +{"spark": "400db173"} +{"spark": "401"} +{"spark": "402"} +{"spark": "403"} +{"spark": "404"} +{"spark": "411"} +{"spark": "412"} +{"spark": "413"} +{"spark": "420"} +{"spark": "500"} +spark-rapids-shim-json-lines ***/ +package com.nvidia.spark.rapids.shims + +import java.io.FileNotFoundException +import java.util.concurrent.ExecutionException + +import com.nvidia.spark.rapids.GpuFileNotFoundException + +import org.apache.spark.sql.catalyst.InternalRow +import org.apache.spark.sql.connector.read.{InputPartition, PartitionReader, PartitionReaderFactory} +import org.apache.spark.sql.execution.datasources.FilePartition +import org.apache.spark.sql.execution.datasources.v2.FileDataSourceV2 +import org.apache.spark.sql.vectorized.ColumnarBatch + +object MissingFileErrorShim { + def wrapReaderFactory(readerFactory: PartitionReaderFactory): PartitionReaderFactory = + new PartitionReaderFactory { + override def createReader(partition: InputPartition): PartitionReader[InternalRow] = + wrapReader(partition, readerFactory.createReader(partition)) + + override def createColumnarReader( + partition: InputPartition): PartitionReader[ColumnarBatch] = + wrapReader(partition, readerFactory.createColumnarReader(partition)) + + override def supportColumnarReads(partition: InputPartition): Boolean = + readerFactory.supportColumnarReads(partition) + } + + private def wrapReader[T]( + partition: InputPartition, + createReader: => PartitionReader[T]): PartitionReader[T] = { + val reader = withStructuredMissingFile(partition)(createReader) + new PartitionReader[T] { + override def next(): Boolean = withStructuredMissingFile(partition)(reader.next()) + + override def get(): T = withStructuredMissingFile(partition)(reader.get()) + + override def close(): Unit = reader.close() + } + } + + private def withStructuredMissingFile[T](partition: InputPartition)(body: => T): T = { + try { + body + } catch { + case error: FileNotFoundException => + throw convertMissingFile(partition, error) + case error: ExecutionException => + error.getCause match { + case cause: FileNotFoundException => throw convertMissingFile(partition, cause) + case _ => throw error + } + } + } + + private def convertMissingFile( + partition: InputPartition, + error: FileNotFoundException): Throwable = { + val (filePath, originalError) = error match { + case GpuFileNotFoundException(path, originalException) => + (Some(path), originalException) + case _ => + (singleFilePath(partition), error) + } + convert(filePath, originalError, includeRefreshHint = false) + } + + private def singleFilePath(partition: InputPartition): Option[String] = { + Option(partition).collect { case filePartition: FilePartition => + SparkShimImpl.getPartitionFiles(filePartition) + }.filter(_.length == 1).map(_.head.filePath.toString) + } + + def convert( + filePath: Option[String], + error: FileNotFoundException, + includeRefreshHint: Boolean): Throwable = filePath match { + case Some(path) => FileDataSourceV2.attachFilePath(path, error) + case None => error + } +} diff --git a/sql-plugin/src/main/spark420/scala/com/nvidia/spark/rapids/shims/GpuBatchScanExec.scala b/sql-plugin/src/main/spark420/scala/com/nvidia/spark/rapids/shims/GpuBatchScanExec.scala index 4cea8ad86ec..e67c42f9251 100644 --- a/sql-plugin/src/main/spark420/scala/com/nvidia/spark/rapids/shims/GpuBatchScanExec.scala +++ b/sql-plugin/src/main/spark420/scala/com/nvidia/spark/rapids/shims/GpuBatchScanExec.scala @@ -144,7 +144,8 @@ case class GpuBatchScanExec( } } - override lazy val readerFactory: PartitionReaderFactory = batch.createReaderFactory() + override lazy val readerFactory: PartitionReaderFactory = + MissingFileErrorShim.wrapReaderFactory(batch.createReaderFactory()) override lazy val inputRDD: RDD[InternalRow] = { scan.metrics = allMetrics @@ -155,7 +156,8 @@ case class GpuBatchScanExec( sparkContext, filteredPartitions.map(_.toSeq), readerFactory, - new Spark42GpuDataSourceCustomMetricsFactory(scanCustomSQLMetrics)) + includeRefreshHint = false, + customMetricsFactory = new Spark42GpuDataSourceCustomMetricsFactory(scanCustomSQLMetrics)) } postDriverMetrics(scan.reportDriverMetrics()) rdd diff --git a/sql-plugin/src/main/spark500/scala/com/nvidia/spark/rapids/shims/GpuBatchScanExec.scala b/sql-plugin/src/main/spark500/scala/com/nvidia/spark/rapids/shims/GpuBatchScanExec.scala index 969c66b7fe4..6a1631a4ae6 100644 --- a/sql-plugin/src/main/spark500/scala/com/nvidia/spark/rapids/shims/GpuBatchScanExec.scala +++ b/sql-plugin/src/main/spark500/scala/com/nvidia/spark/rapids/shims/GpuBatchScanExec.scala @@ -97,7 +97,8 @@ case class GpuBatchScanExec( outputPartitioning, inputPartitions) - override lazy val readerFactory: PartitionReaderFactory = batch.createReaderFactory() + override lazy val readerFactory: PartitionReaderFactory = + MissingFileErrorShim.wrapReaderFactory(batch.createReaderFactory()) override lazy val inputRDD: RDD[InternalRow] = { scan.metrics = allMetrics @@ -108,7 +109,8 @@ case class GpuBatchScanExec( sparkContext, filteredPartitions.map(_.toSeq), readerFactory, - new Spark42GpuDataSourceCustomMetricsFactory(scanCustomSQLMetrics)) + includeRefreshHint = false, + customMetricsFactory = new Spark42GpuDataSourceCustomMetricsFactory(scanCustomSQLMetrics)) } postDriverMetrics(scan.reportDriverMetrics()) rdd diff --git a/tests/src/test/scala/com/nvidia/spark/rapids/FileSystemBytesReadTrackerSuite.scala b/tests/src/test/scala/com/nvidia/spark/rapids/FileSystemBytesReadTrackerSuite.scala index 7f1325af3e7..cb6136f4227 100644 --- a/tests/src/test/scala/com/nvidia/spark/rapids/FileSystemBytesReadTrackerSuite.scala +++ b/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.{GpuDataSourceCustomMetrics, @@ -25,7 +27,7 @@ import org.mockito.Mockito.{verify, when} import org.scalatest.funsuite.AnyFunSuite import org.scalatestplus.mockito.MockitoSugar -import org.apache.spark.SparkContext +import org.apache.spark.{SparkContext, SparkException} import org.apache.spark.sql.connector.read.{InputPartition, PartitionReader, PartitionReaderFactory} import org.apache.spark.sql.rapids.execution.TrampolineUtil import org.apache.spark.sql.rapids.metrics.source.MockTaskContext @@ -118,7 +120,8 @@ class FileSystemBytesReadTrackerSuite extends AnyFunSuite with MockitoSugar { override def supportColumnarReads(partition: InputPartition): Boolean = true } - val rdd = GpuDataSourceRDD(mock[SparkContext], Seq(inputPartition), factory) + val rdd = GpuDataSourceRDD( + mock[SparkContext], Seq(inputPartition), factory, includeRefreshHint = false) val iterator = rdd.compute(rdd.partitions.head, context) assert(iterator.hasNext) @@ -129,6 +132,72 @@ class FileSystemBytesReadTrackerSuite extends AnyFunSuite with MockitoSugar { } } + private val missingFilePath = "file:/missing%20ORC.orc" + + Seq( + ("direct V2", false, + () => GpuFileNotFoundException( + missingFilePath, new FileNotFoundException("missing ORC file"))), + ("wrapped V1", true, + () => new ExecutionException(GpuFileNotFoundException( + missingFilePath, 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 = includeRefreshHint) + val iterator = rdd.compute(rdd.partitions.head, context) + + assert(iterator.hasNext) + val error = intercept[Exception](iterator.next()) + if (VersionUtils.isSpark400OrLater) { + val sparkError = error.asInstanceOf[SparkException] + val condition = sparkError.getClass.getMethod("getCondition").invoke(sparkError) + val parameters = sparkError.getClass.getMethod("getMessageParameters") + .invoke(sparkError).asInstanceOf[java.util.Map[String, String]] + assert(condition === "FAILED_READ_FILE.FILE_NOT_EXIST") + assert(parameters.get("path") === missingFilePath) + } else { + val fileError = error.asInstanceOf[FileNotFoundException] + assert(fileError.getMessage.contains("recreating the Dataset/DataFrame involved")) + assert(fileError.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 {} @@ -151,7 +220,8 @@ class FileSystemBytesReadTrackerSuite extends AnyFunSuite with MockitoSugar { override def supportColumnarReads(partition: InputPartition): Boolean = true } - val rdd = GpuDataSourceRDD(mock[SparkContext], Seq(inputPartition), factory) + val rdd = GpuDataSourceRDD( + mock[SparkContext], Seq(inputPartition), factory, includeRefreshHint = false) val iterator = rdd.compute(rdd.partitions.head, context) assert(iterator.hasNext) @@ -190,7 +260,8 @@ class FileSystemBytesReadTrackerSuite extends AnyFunSuite with MockitoSugar { } } val rdd = new GpuDataSourceRDD( - mock[SparkContext], Seq(Seq(inputPartition)), readerFactory, customMetricsFactory) + mock[SparkContext], Seq(Seq(inputPartition)), readerFactory, + includeRefreshHint = false, customMetricsFactory = customMetricsFactory) val iterator = rdd.compute(rdd.partitions.head, context) assert(iterator.hasNext) @@ -230,7 +301,8 @@ class FileSystemBytesReadTrackerSuite extends AnyFunSuite with MockitoSugar { override def supportColumnarReads(partition: InputPartition): Boolean = true } - GpuDataSourceRDD(mock[SparkContext], Seq(inputPartition), factory) + GpuDataSourceRDD( + mock[SparkContext], Seq(inputPartition), factory, includeRefreshHint = false) } val firstRdd = newRdd(10L) diff --git a/tests/src/test/spark330/scala/org/apache/spark/sql/rapids/suites/RapidsMetadataCacheSuite.scala b/tests/src/test/spark330/scala/org/apache/spark/sql/rapids/suites/RapidsMetadataCacheSuite.scala index 9d7643daac6..43c18188a5a 100644 --- a/tests/src/test/spark330/scala/org/apache/spark/sql/rapids/suites/RapidsMetadataCacheSuite.scala +++ b/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 +} diff --git a/tests/src/test/spark330/scala/org/apache/spark/sql/rapids/utils/RapidsTestSettings.scala b/tests/src/test/spark330/scala/org/apache/spark/sql/rapids/utils/RapidsTestSettings.scala index 1672bc48ab7..2a976e7cd99 100644 --- a/tests/src/test/spark330/scala/org/apache/spark/sql/rapids/utils/RapidsTestSettings.scala +++ b/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")) diff --git a/tests/src/test/spark400/scala/com/nvidia/spark/rapids/MissingFileStructuredErrorSuite.scala b/tests/src/test/spark400/scala/com/nvidia/spark/rapids/MissingFileStructuredErrorSuite.scala new file mode 100644 index 00000000000..fa2cf02e4c3 --- /dev/null +++ b/tests/src/test/spark400/scala/com/nvidia/spark/rapids/MissingFileStructuredErrorSuite.scala @@ -0,0 +1,106 @@ +/* + * 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": "400"} +{"spark": "400db173"} +{"spark": "401"} +{"spark": "402"} +{"spark": "403"} +{"spark": "404"} +{"spark": "411"} +{"spark": "412"} +{"spark": "413"} +{"spark": "420"} +spark-rapids-shim-json-lines ***/ +package com.nvidia.spark.rapids + +import java.io.File + +import com.nvidia.spark.rapids.shims.GpuBatchScanExec + +import org.apache.spark.{SparkConf, SparkException, SparkThrowable} +import org.apache.spark.sql.SparkSession +import org.apache.spark.sql.internal.SQLConf +import org.apache.spark.sql.rapids.GpuFileSourceScanExec + +class MissingFileStructuredErrorSuite extends SparkQueryCompareTestSuite { + private val missingFileCondition = "FAILED_READ_FILE.FILE_NOT_EXIST" + + private def deleteOneDataFile(directory: File): File = { + val file = directory.listFiles().find { candidate => + !candidate.getName.startsWith("_") && !candidate.getName.startsWith(".") + }.getOrElse(fail(s"No data file found in $directory")) + assert(file.delete(), s"Failed to delete $file") + file + } + + private def structuredMissingFile(error: Throwable): SparkThrowable = { + Iterator.iterate(error)(_.getCause).takeWhile(_ != null).collectFirst { + case sparkError: SparkThrowable if sparkError.getCondition == missingFileCondition => + sparkError + }.getOrElse(fail(s"No $missingFileCondition error found in $error")) + } + + private def readAfterDeletingPlannedFile( + spark: SparkSession, + useV1: Boolean, + verifyGpuPlan: Boolean): String = { + 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) + + if (verifyGpuPlan) { + val plan = df.queryExecution.executedPlan + val hasExpectedGpuScan = if (useV1) { + plan.find(_.isInstanceOf[GpuFileSourceScanExec]).nonEmpty + } else { + plan.find(_.isInstanceOf[GpuBatchScanExec]).nonEmpty + } + assert(hasExpectedGpuScan, s"Expected a GPU ${if (useV1) "V1" else "V2"} scan:\n$plan") + } + + val deletedFile = deleteOneDataFile(location) + val error = structuredMissingFile(intercept[SparkException](df.count())) + val expectedPath = deletedFile.toPath.toUri.toString + assert(error.getMessageParameters.get("path") == expectedPath) + error.getCondition + } + } + + Seq(("V1", true, "orc"), ("V2", false, "")).foreach { + case (sourceName, useV1, v1Sources) => + Seq(RapidsReaderType.COALESCING, RapidsReaderType.MULTITHREADED).foreach { readerType => + test(s"Spark 4 missing-file structured error parity - $sourceName - $readerType") { + val conf = new SparkConf() + .set(SQLConf.USE_V1_SOURCE_LIST.key, v1Sources) + .set(SQLConf.IGNORE_MISSING_FILES.key, "false") + .set(RapidsConf.ORC_READER_TYPE.key, readerType.toString) + + val cpuCondition = withCpuSparkSession( + readAfterDeletingPlannedFile(_, useV1, verifyGpuPlan = false), conf) + val gpuCondition = withGpuSparkSession( + readAfterDeletingPlannedFile(_, useV1, verifyGpuPlan = true), conf) + + assert(gpuCondition == cpuCondition) + assert(gpuCondition == missingFileCondition) + } + } + } +}