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. The per-file reader fills in the
+    /// original Spark path before the JVM wraps this in cannotReadFilesError.
+    #[error("long overflow")]
+    ParquetTimestampOverflow { file_path: String },
+
     /// 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",
@@ -611,6 +622,9 @@ impl SparkError {
                     "sparkType": spark_type,
                 })
             }
+            SparkError::ParquetTimestampOverflow { file_path } => {
+                serde_json::json!({ "filePath": file_path })
+            }
             SparkError::CannotReadFile { file_path, message } => {
                 serde_json::json!({
                     "filePath": file_path,
@@ -635,6 +649,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 +730,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 +758,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 +832,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 +976,38 @@ mod tests {
         assert!(json.contains("\"errorClass\":\"REMAINDER_BY_ZERO\""));
     }
 
+    #[test]
+    fn test_long_overflow_json() {
+        for (error, error_type, exception_class, params) in [
+            (
+                SparkError::LongOverflow,
+                "LongOverflow",
+                "java/lang/ArithmeticException",
+                serde_json::json!({}),
+            ),
+            (
+                SparkError::ParquetTimestampOverflow {
+                    file_path: "file:///bad%20timestamp.parquet".to_string(),
+                },
+                "ParquetTimestampOverflow",
+                "org/apache/spark/SparkException",
+                serde_json::json!({ "filePath": "file:///bad%20timestamp.parquet" }),
+            ),
+        ] {
+            let parsed: serde_json::Value = serde_json::from_str(&error.to_json()).unwrap();
+            assert_eq!(
+                parsed,
+                serde_json::json!({
+                    "errorType": error_type,
+                    "errorClass": "",
+                    "params": 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/execution/operators/dynamic_filter.rs

@@ -54,6 +54,7 @@ use datafusion::physical_plan::{
 use futures::StreamExt;
 
 use super::CometFilterExec;
+use crate::parquet::file_error_context::ParquetErrorContext;
 
 /// A task-local consumer of DataFusion's build-side runtime filter.
 #[derive(Debug)]
@@ -293,7 +294,11 @@ fn try_attach_parquet_reader_filter(
         );
         return Ok(None);
     };
-    if scan.downcast_to_file_source::<ParquetSource>().is_none() {
+    if scan.downcast_to_file_source::<ParquetSource>().is_none()
+        && scan
+            .downcast_to_file_source::<ParquetErrorContext>()
+            .is_none()
+    {
         log::debug!("Join dynamic filter reader pushdown skipped: probe is not Parquet");
         return Ok(None);
     }

native/core/src/execution/planner.rs

@@ -472,6 +472,11 @@ impl PhysicalPlanner {
             let url = normalize_object_store_url(&file.file_path, object_store_options)?.url;
             let path = Path::from_url_path(url.path()).map_err(|e| GeneralError(e.to_string()))?;
             partitioned_file.object_meta.location = path;
+            partitioned_file
+                .extensions
+                .insert(crate::parquet::file_error_context::SparkFilePath(
+                    Arc::from(file.file_path.as_str()),
+                ));
 
             // Process partition values
             // Create an empty input schema for partition values because they are all literals.