diff --git a/native/common/src/error.rs b/native/common/src/error.rs index 41773237cba..1bb53fe1b81 100644 --- a/native/common/src/error.rs +++ b/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 { diff --git a/native/core/src/parquet/parquet_support.rs b/native/core/src/parquet/parquet_support.rs index c815944c468..efd4272f46f 100644 --- a/native/core/src/parquet/parquet_support.rs +++ b/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), µs_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::(), 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::(), Some(SparkError::ParquetTimestampOverflow)))); let converted = parquet_convert_array(strukt, &target, &unchecked_options) .expect("filtered nested overflow must not error"); let converted_child = Arc::clone( diff --git a/spark/src/main/scala/org/apache/comet/CometExecIterator.scala b/spark/src/main/scala/org/apache/comet/CometExecIterator.scala index e2c132904d5..5eba16b0a9d 100644 --- a/spark/src/main/scala/org/apache/comet/CometExecIterator.scala +++ b/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 diff --git a/spark/src/main/scala/org/apache/comet/SparkErrorConverter.scala b/spark/src/main/scala/org/apache/comet/SparkErrorConverter.scala index a6bc21aca71..a9c63a9aa6b 100644 --- a/spark/src/main/scala/org/apache/comet/SparkErrorConverter.scala +++ b/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 diff --git a/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimSparkErrorConverter.scala b/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimSparkErrorConverter.scala index dcb73971901..0be240320c6 100644 --- a/spark/src/main/spark-3.4/org/apache/spark/sql/comet/shims/ShimSparkErrorConverter.scala +++ b/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 diff --git a/spark/src/main/spark-3.5/org/apache/spark/sql/comet/shims/ShimSparkErrorConverter.scala b/spark/src/main/spark-3.5/org/apache/spark/sql/comet/shims/ShimSparkErrorConverter.scala index 571eaf05547..26964e7bc7f 100644 --- a/spark/src/main/spark-3.5/org/apache/spark/sql/comet/shims/ShimSparkErrorConverter.scala +++ b/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 diff --git a/spark/src/main/spark-4.x/org/apache/spark/sql/comet/shims/ShimSparkErrorConverter.scala b/spark/src/main/spark-4.x/org/apache/spark/sql/comet/shims/ShimSparkErrorConverter.scala index 7397745885c..9181f963834 100644 --- a/spark/src/main/spark-4.x/org/apache/spark/sql/comet/shims/ShimSparkErrorConverter.scala +++ b/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 diff --git a/spark/src/test/scala/org/apache/comet/SparkErrorConverterSuite.scala b/spark/src/test/scala/org/apache/comet/SparkErrorConverterSuite.scala index 3b81a76ead5..04cf42e6c8c 100644 --- a/spark/src/test/scala/org/apache/comet/SparkErrorConverterSuite.scala +++ b/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( diff --git a/spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala b/spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala index d412918b572..888cfe1a612 100644 --- a/spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala +++ b/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") } } }