fix: preserve Spark errors for Parquet timestamp overflow

native/common/src/error.rs

@@ -69,6 +69,10 @@ pub enum SparkError {
     #[error("[ARITHMETIC_OVERFLOW] {from_type} overflow. If necessary set \"spark.sql.ansi.enabled\" to \"false\" to bypass this error.")]
     ArithmeticOverflow { from_type: String },
 
+    // Spark's checked date/timestamp conversions throw this even with ANSI disabled.
+    #[error("long overflow")]
+    LongOverflow,
+
     #[error("[ARITHMETIC_OVERFLOW] Overflow in integral divide. Use 'try_divide' to tolerate overflow and return NULL instead. If necessary set \"spark.sql.ansi.enabled\" to \"false\" to bypass this error.")]
     IntegralDivideOverflow,
 
@@ -247,6 +251,11 @@ pub enum SparkError {
         spark_type: String,
     },
 
+    /// Overflow in Parquet's millis-to-micros conversion. Unlike a cast's LongOverflow, the JVM
+    /// wraps this in cannotReadFilesError, using the per-task file list for the path.
+    #[error("long overflow")]
+    ParquetTimestampOverflow,
+
     /// A per-file read failure (corrupt footer/page, truncated/empty file, deleted file) raised by
     /// the native parquet reader / object_store. Classified by typed `DataFusionError` variant (no
     /// message matching) and translated by the JVM shim into Spark's `FAILED_READ_FILE`
@@ -308,6 +317,7 @@ impl SparkError {
             SparkError::CastOverFlow { .. } => "CastOverFlow",
             SparkError::CannotParseDecimal => "CannotParseDecimal",
             SparkError::ArithmeticOverflow { .. } => "ArithmeticOverflow",
+            SparkError::LongOverflow => "LongOverflow",
             SparkError::IntegralDivideOverflow => "IntegralDivideOverflow",
             SparkError::DecimalSumOverflow { .. } => "DecimalSumOverflow",
             SparkError::DivideByZero => "DivideByZero",
@@ -349,6 +359,7 @@ impl SparkError {
             SparkError::DuplicateFieldByFieldId { .. } => "DuplicateFieldByFieldId",
             SparkError::ParquetMissingFieldIds => "ParquetMissingFieldIds",
             SparkError::ParquetSchemaConvert { .. } => "ParquetSchemaConvert",
+            SparkError::ParquetTimestampOverflow => "ParquetTimestampOverflow",
             SparkError::CannotReadFile { .. } => "CannotReadFile",
             SparkError::Arrow(_) => "Arrow",
             SparkError::Internal(_) => "Internal",
@@ -635,6 +646,8 @@ impl SparkError {
     /// Returns the appropriate Spark exception class for this error
     pub fn exception_class(&self) -> &'static str {
         match self {
+            SparkError::LongOverflow => "java/lang/ArithmeticException",
+
             // ArithmeticException
             SparkError::DivideByZero
             | SparkError::RemainderByZero
@@ -714,9 +727,10 @@ impl SparkError {
                 "org/apache/spark/sql/execution/datasources/SchemaColumnConvertNotSupportedException"
             }
 
-            // CannotReadFile - converted to a FAILED_READ_FILE SparkException by the shim
-            // (QueryExecutionErrors.cannotReadFilesError).
-            SparkError::CannotReadFile { .. } => "org/apache/spark/SparkException",
+            // File-read failures are wrapped by QueryExecutionErrors.cannotReadFilesError.
+            SparkError::CannotReadFile { .. } | SparkError::ParquetTimestampOverflow => {
+                "org/apache/spark/SparkException"
+            }
 
             // Generic errors
             SparkError::Arrow(_) | SparkError::Internal(_) => "org/apache/spark/SparkException",
@@ -741,6 +755,7 @@ impl SparkError {
             SparkError::RemainderByZero => Some("REMAINDER_BY_ZERO"),
             SparkError::IntervalDividedByZero => Some("INTERVAL_DIVIDED_BY_ZERO"),
             SparkError::ArithmeticOverflow { .. } => Some("ARITHMETIC_OVERFLOW"),
+            SparkError::LongOverflow => None,
             SparkError::IntegralDivideOverflow => Some("ARITHMETIC_OVERFLOW"),
             SparkError::DecimalSumOverflow { .. } => Some("ARITHMETIC_OVERFLOW"),
             SparkError::BinaryArithmeticOverflow { .. } => Some("BINARY_ARITHMETIC_OVERFLOW"),
@@ -814,9 +829,8 @@ impl SparkError {
             // SparkException error class, so no error class is exposed here.
             SparkError::ParquetSchemaConvert { .. } => None,
 
-            // CannotReadFile — the JVM shim wraps it via cannotReadFilesError, which supplies the
-            // FAILED_READ_FILE error class, so none is exposed here.
-            SparkError::CannotReadFile { .. } => None,
+            // The JVM's cannotReadFilesError supplies the version-appropriate error class.
+            SparkError::CannotReadFile { .. } | SparkError::ParquetTimestampOverflow => None,
 
             // Generic errors (no error class)
             SparkError::Arrow(_) | SparkError::Internal(_) => None,
@@ -959,6 +973,34 @@ mod tests {
         assert!(json.contains("\"errorClass\":\"REMAINDER_BY_ZERO\""));
     }
 
+    #[test]
+    fn test_long_overflow_json() {
+        for (error, error_type, exception_class) in [
+            (
+                SparkError::LongOverflow,
+                "LongOverflow",
+                "java/lang/ArithmeticException",
+            ),
+            (
+                SparkError::ParquetTimestampOverflow,
+                "ParquetTimestampOverflow",
+                "org/apache/spark/SparkException",
+            ),
+        ] {
+            let parsed: serde_json::Value = serde_json::from_str(&error.to_json()).unwrap();
+            assert_eq!(
+                parsed,
+                serde_json::json!({
+                    "errorType": error_type,
+                    "errorClass": "",
+                    "params": {},
+                })
+            );
+            assert_eq!(error.exception_class(), exception_class);
+            assert_eq!(error.to_string(), "long overflow");
+        }
+    }
+
     #[test]
     fn test_binary_overflow_json() {
         let error = SparkError::BinaryArithmeticOverflow {

native/core/src/parquet/parquet_support.rs

@@ -26,7 +26,7 @@ use arrow::datatypes::{FieldRef, Fields};
 use arrow::{
     array::{
         cast::AsArray, new_null_array, types::TimestampMicrosecondType,
-        types::TimestampMillisecondType, Array, ArrayRef, ArrowNativeTypeOp, StructArray,
+        types::TimestampMillisecondType, Array, ArrayRef, StructArray,
     },
     compute::{cast_with_options, CastOptions},
     datatypes::{DataType, TimeUnit},
@@ -269,7 +269,9 @@ fn parquet_convert_array_impl(
             // Restore the original child validity: required fields must remain non-null.
             let micros = arrow::array::TimestampMillisecondArray::new(
                 millis.values().clone(), visible)
-                .try_unary::<_, TimestampMicrosecondType, _>(|value| value.mul_checked(1_000))?;
+                .try_unary::<_, TimestampMicrosecondType, _>(|value| {
+                    value.checked_mul(1_000).ok_or(SparkError::ParquetTimestampOverflow)
+                })?;
             let micros = arrow::array::TimestampMicrosecondArray::new(
                 micros.values().clone(), millis.nulls().cloned())
                 .with_timezone_opt(target_tz.clone());
@@ -1325,6 +1327,8 @@ mod tests {
         use crate::parquet::parquet_support::{parquet_convert_array, SparkParquetOptions};
         use arrow::array::{Array, ArrayRef, StructArray, TimestampMillisecondArray};
         use arrow::datatypes::{DataType, Field, Fields, TimeUnit};
+        use datafusion::error::DataFusionError;
+        use datafusion_comet_common::SparkError;
         use datafusion_comet_spark_expr::EvalMode;
         use std::sync::Arc;
 
@@ -1339,10 +1343,8 @@ mod tests {
         // Top-level: checked, matching Spark's `millisToMicros` (`Math.multiplyExact`).
         let err = parquet_convert_array(Arc::clone(&millis), &micros_type, &options)
             .expect_err("top-level overflow must error");
-        assert!(
-            err.to_string().to_lowercase().contains("overflow"),
-            "unexpected error: {err}"
-        );
+        assert!(matches!(err, DataFusionError::External(ref source)
+            if matches!(source.downcast_ref::<SparkError>(), Some(SparkError::ParquetTimestampOverflow))));
 
         // Filtered scans disable checked conversion because Spark may prune values before
         // conversion through paths DataFusion cannot fully mirror.
@@ -1370,7 +1372,10 @@ mod tests {
             micros_type.clone(),
             true,
         ))]));
-        assert!(parquet_convert_array(Arc::clone(&strukt), &target, &options).is_err());
+        let err = parquet_convert_array(Arc::clone(&strukt), &target, &options)
+            .expect_err("nested overflow must error");
+        assert!(matches!(err, DataFusionError::External(ref source)
+            if matches!(source.downcast_ref::<SparkError>(), Some(SparkError::ParquetTimestampOverflow))));
         let converted = parquet_convert_array(strukt, &target, &unchecked_options)
             .expect("filtered nested overflow must not error");
         let converted_child = Arc::clone(

spark/src/main/scala/org/apache/comet/CometExecIterator.scala

@@ -21,6 +21,8 @@ package org.apache.comet
 
 import java.lang.management.ManagementFactory
 
+import scala.jdk.CollectionConverters._
+
 import org.apache.arrow.c.ArrowArrayStream
 import org.apache.hadoop.conf.Configuration
 import org.apache.spark._
@@ -37,6 +39,7 @@ import org.apache.comet.Tracing.withTrace
 import org.apache.comet.exceptions.CometQueryExecutionException
 import org.apache.comet.parquet.CometFileKeyUnwrapper
 import org.apache.comet.serde.Config.ConfigMap
+import org.apache.comet.serde.OperatorOuterClass.Operator
 import org.apache.comet.shuffle.ShufflePartitionPusher
 import org.apache.comet.vector.NativeUtil
 
@@ -215,7 +218,20 @@ class CometExecIterator(
       // Handle CometQueryExecutionException with JSON payload first
       case e: CometQueryExecutionException =>
         logError(s"Native execution for task $taskAttemptId failed", e)
-        throw SparkErrorConverter.convertToSparkException(e, taskFilePaths)
+        // Fused scans (including shuffle writers) carry paths in the injected task plan, but
+        // may not supply taskFilePaths. Read that metadata only when execution fails.
+        def scanPaths(op: Operator): Seq[String] = {
+          val paths = op.getNativeScan.getFilePartition.getPartitionedFileList.asScala
+            .map(_.getFilePath)
+            .toSeq
+          paths ++ op.getChildrenList.asScala.flatMap(scanPaths)
+        }
+        val filePaths =
+          if (taskFilePaths.nonEmpty) taskFilePaths
+          else {
+            scanPaths(Operator.parseFrom(protobufQueryPlan)).distinct
+          }
+        throw SparkErrorConverter.convertToSparkException(e, filePaths)
 
       case e: CometNativeException =>
         // it is generally considered bad practice to log and then rethrow an

spark/src/main/scala/org/apache/comet/SparkErrorConverter.scala

@@ -86,11 +86,10 @@ object SparkErrorConverter extends ShimSparkErrorConverter {
     val json = parse(e.getMessage)
     val errorJson = json.extract[ErrorJson]
     val rawParams = errorJson.params.getOrElse(Map.empty)
-    // CannotReadFile carries the offending file path natively only for the object_store NotFound
-    // case; for corrupt/truncated parquet the native error has no path, so fall back to the
-    // per-task file list threaded in from CometExecIterator.
+    // File-read errors without a native path use the per-task file list from CometExecIterator.
     val params =
-      if (errorJson.errorType == "CannotReadFile"
+      if ((errorJson.errorType == "CannotReadFile" ||
+          errorJson.errorType == "ParquetTimestampOverflow")
         && rawParams.get("filePath").forall(p => p == null || p.toString.isEmpty)
         && taskFilePaths.nonEmpty) {
         rawParams + ("filePath" -> taskFilePaths.mkString(","))
@@ -117,8 +116,13 @@ object SparkErrorConverter extends ShimSparkErrorConverter {
 
     val summary: String = errorJson.summary.getOrElse("")
 
-    // Delegate to version-specific shim - let conversion exceptions propagate
-    val optEx = convertErrorType(errorJson.errorType, errorClass, params, sparkContext, summary)
+    // Math.multiplyExact throws a plain JVM exception in every Spark version, without an
+    // ANSI error class or configuration advice. Delegate other errors to the version-specific shim.
+    val optEx = if (errorJson.errorType == "LongOverflow") {
+      Some(new ArithmeticException("long overflow"))
+    } else {
+      convertErrorType(errorJson.errorType, errorClass, params, sparkContext, summary)
+    }
     optEx match {
       case Some(exception) =>
         // successfully converted - return the proper typed exception

spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimSparkErrorConverter.scala

@@ -373,6 +373,12 @@ trait ShimSparkErrorConverter {
           QueryExecutionErrors.readCurrentFileNotFoundError(
             new FileNotFoundException(s"File $path does not exist")))
 
+      case "ParquetTimestampOverflow" =>
+        val filePath = params.get("filePath").map(_.toString).getOrElse("")
+        Some(
+          QueryExecutionErrors
+            .cannotReadFilesError(new ArithmeticException("long overflow"), filePath))
+
       case "CannotReadFile" =>
         // A per-file read failure of a readable-but-broken file (corrupt/truncated parquet,
         // object_store, IO) classified by typed DataFusionError variant on the native side. Wrap

spark/src/main/spark-3.5/org/apache/spark/sql/comet/shims/ShimSparkErrorConverter.scala

@@ -369,6 +369,12 @@ trait ShimSparkErrorConverter {
           QueryExecutionErrors.readCurrentFileNotFoundError(
             new FileNotFoundException(s"File $path does not exist")))
 
+      case "ParquetTimestampOverflow" =>
+        val filePath = params.get("filePath").map(_.toString).getOrElse("")
+        Some(
+          QueryExecutionErrors
+            .cannotReadFilesError(new ArithmeticException("long overflow"), filePath))
+
       case "CannotReadFile" =>
         // A per-file read failure (corrupt/truncated/deleted parquet, object_store, IO) classified
         // by typed DataFusionError variant on the native side. Wrap in the FAILED_READ_FILE

spark/src/main/spark-4.x/org/apache/spark/sql/comet/shims/ShimSparkErrorConverter.scala

@@ -386,6 +386,12 @@ trait ShimSparkErrorConverter {
           QueryExecutionErrors
             .fileNotExistError(path, new FileNotFoundException(s"File $path does not exist")))
 
+      case "ParquetTimestampOverflow" =>
+        val filePath = params.get("filePath").map(_.toString).getOrElse("")
+        Some(
+          QueryExecutionErrors
+            .cannotReadFilesError(new ArithmeticException("long overflow"), filePath))
+
       case "CannotReadFile" =>
         // A per-file read failure (corrupt/truncated/deleted parquet, object_store, IO) classified
         // by typed DataFusionError variant on the native side. Wrap in the FAILED_READ_FILE

spark/src/test/scala/org/apache/comet/SparkErrorConverterSuite.scala

@@ -21,8 +21,40 @@ package org.apache.comet
 
 import org.scalatest.funsuite.AnyFunSuite
 
+import org.apache.spark.SparkException
+
+import org.apache.comet.CometSparkSessionExtensions.isSpark40Plus
+
 class SparkErrorConverterSuite extends AnyFunSuite {
 
+  test("LongOverflow converts to a plain ArithmeticException") {
+    val json = """{"errorType":"LongOverflow","errorClass":"","params":{}}"""
+    // A cast above a scan can have task file paths too; it must remain a plain exception.
+    Seq(Seq.empty[String], Seq("file:/tmp/data/part-0.parquet")).foreach { paths =>
+      val ex = SparkErrorConverter.convertToSparkException(
+        new org.apache.comet.exceptions.CometQueryExecutionException(json),
+        taskFilePaths = paths)
+      assert(ex.getClass == classOf[ArithmeticException])
+      assert(ex.getMessage == "long overflow")
+    }
+  }
+
+  test("ParquetTimestampOverflow wraps the arithmetic cause in a file-read SparkException") {
+    val json = """{"errorType":"ParquetTimestampOverflow","errorClass":"","params":{}}"""
+    val path = "file:/tmp/data/part-0.parquet"
+    val ex = SparkErrorConverter.convertToSparkException(
+      new org.apache.comet.exceptions.CometQueryExecutionException(json),
+      taskFilePaths = Seq(path))
+    assert(ex.getClass == classOf[SparkException])
+    val error = ex.asInstanceOf[SparkException]
+    val errorClass =
+      if (isSpark40Plus) "FAILED_READ_FILE.NO_HINT" else "_LEGACY_ERROR_TEMP_2064"
+    assert(error.getErrorClass == errorClass)
+    assert(error.getMessageParameters.get("path") == path)
+    assert(error.getCause.getClass == classOf[ArithmeticException])
+    assert(error.getCause.getMessage == "long overflow")
+  }
+
   test("CannotReadFile converts to a FAILED_READ_FILE SparkException naming the file") {
     val ex = SparkErrorConverter
       .convertErrorType(

spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala

@@ -271,11 +271,8 @@ abstract class ParquetReadSuite extends CometTestBase {
     // https://github.com/apache/spark/blob/v4.2.0/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/ParquetVectorUpdaterFactory.java#L800-L833
     // Matches Spark's positive and negative overflow cases:
     // https://github.com/apache/spark/blob/v4.2.0/sql/core/src/test/resources/sql-tests/inputs/timestamp.sql#L74-L83
-    def isOverflow(error: Throwable): Boolean =
-      Iterator
-        .iterate(error)(_.getCause)
-        .takeWhile(_ != null)
-        .exists(cause => Option(cause.getMessage).exists(_.toLowerCase.contains("overflow")))
+    val errorClass =
+      if (isSpark40Plus) "FAILED_READ_FILE.NO_HINT" else "_LEGACY_ERROR_TEMP_2064"
 
     Seq(false, true).foreach { dictionaryEnabled =>
       Seq(92233720368547758L, -92233720368547758L).foreach { millis =>
@@ -340,14 +337,17 @@ abstract class ParquetReadSuite extends CometTestBase {
           Seq(false, true).foreach { ansiEnabled =>
             withSQLConf(SQLConf.ANSI_ENABLED.key -> ansiEnabled.toString) {
               readParquetFile(path.toString) { df =>
-                Seq("ts", "ts_ntz", "s", "s.ts", "s.ts_ntz", "a", "m").foreach { column =>
-                  val selected = df.select(column)
+                val queries = Seq("ts", "ts_ntz", "s", "s.ts", "s.ts_ntz", "a", "m")
+                  .map(column => df.select(column)) :+ df.select("ts").repartition(1)
+                queries.foreach { selected =>
                   assert(collect(selected.queryExecution.executedPlan) {
                     case _: CometNativeScanExec => true
                   }.nonEmpty)
 
-                  val (sparkError, cometError) = checkSparkAnswerMaybeThrows(selected)
-                  assert(Seq(sparkError, cometError).forall(_.exists(isOverflow)))
+                  val error = checkSparkError(selected, errorClass)
+                  assert(new Path(error.getMessageParameters.get("path")) == path)
+                  assert(error.getCause.getClass == classOf[ArithmeticException])
+                  assert(error.getCause.getMessage == "long overflow")
                 }
               }
             }