diff --git a/native/common/src/error.rs b/native/common/src/error.rs index 41773237cba..623d2b6deaf 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. 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 { diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs index 60c206617e3..88d7270f99b 100644 --- a/native/core/src/execution/planner.rs +++ b/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. diff --git a/native/core/src/parquet/file_error_context.rs b/native/core/src/parquet/file_error_context.rs new file mode 100644 index 00000000000..c1209ae9c92 --- /dev/null +++ b/native/core/src/parquet/file_error_context.rs @@ -0,0 +1,297 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you 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. + +//! Preserve the file identity across Parquet's asynchronous planning and decoding. +//! +//! The schema adapter receives no file identity. Capture it when planning a file and retain it +//! through that file's pending planners and streams. Delegate source operations to preserve +//! Parquet's projection, filter, and sort pushdown. + +use arrow::array::RecordBatch; +use datafusion::common::{ + config::ConfigOptions, tree_node::TreeNodeRecursion, DataFusionError, Result, +}; +use datafusion::datasource::physical_plan::{FileScanConfig, FileSource}; +use datafusion::physical_expr::{ + projection::ProjectionExprs, EquivalenceProperties, LexOrdering, PhysicalExpr, PhysicalSortExpr, +}; +use datafusion::physical_plan::{ + filter_pushdown::FilterPushdownPropagation, metrics::ExecutionPlanMetricsSet, + DisplayFormatType, SortOrderPushdownResult, +}; +use datafusion_comet_common::SparkError; +use datafusion_datasource::{ + file_stream::FileOpener, + morsel::{Morsel, MorselPlan, MorselPlanner, Morselizer}, + PartitionedFile, TableSchema, +}; +use futures::{stream::BoxStream, StreamExt, TryStreamExt}; +use object_store::ObjectStore; +use std::{fmt, sync::Arc}; + +/// The original URL-encoded Spark path, before object-store aliases and paths are normalized. +#[derive(Debug, Clone)] +pub(crate) struct SparkFilePath(pub Arc); + +/// Delegate scan planning and pushdown to ParquetSource, wrapping only its per-file work. +pub(crate) struct ParquetErrorContext(Arc); + +impl ParquetErrorContext { + pub(crate) fn wrap(source: Arc) -> Arc { + Arc::new(Self(source)) + } +} + +impl FileSource for ParquetErrorContext { + fn create_file_opener( + &self, + store: Arc, + config: &FileScanConfig, + partition: usize, + ) -> Result> { + self.0.create_file_opener(store, config, partition) + } + + fn create_morselizer( + &self, + store: Arc, + config: &FileScanConfig, + partition: usize, + ) -> Result> { + Ok(Box::new(FileContextMorselizer( + self.0.create_morselizer(store, config, partition)?, + ))) + } + + fn table_schema(&self) -> &TableSchema { + self.0.table_schema() + } + + fn with_batch_size(&self, batch_size: usize) -> Arc { + Self::wrap(self.0.with_batch_size(batch_size)) + } + + fn filter(&self) -> Option> { + self.0.filter() + } + + fn projection(&self) -> Option<&ProjectionExprs> { + self.0.projection() + } + + fn metrics(&self) -> &ExecutionPlanMetricsSet { + self.0.metrics() + } + + fn file_type(&self) -> &str { + self.0.file_type() + } + + fn fmt_extra(&self, t: DisplayFormatType, f: &mut fmt::Formatter) -> fmt::Result { + self.0.fmt_extra(t, f) + } + + fn supports_repartitioning(&self) -> bool { + self.0.supports_repartitioning() + } + + fn repartitioned( + &self, + target_partitions: usize, + min_size: usize, + ordering: Option, + config: &FileScanConfig, + ) -> Result> { + self.0 + .repartitioned(target_partitions, min_size, ordering, config) + } + + fn try_pushdown_filters( + &self, + filters: Vec>, + config: &ConfigOptions, + ) -> Result>> { + let mut result = self.0.try_pushdown_filters(filters, config)?; + result.updated_node = result.updated_node.map(Self::wrap); + Ok(result) + } + + fn try_pushdown_sort( + &self, + order: &[PhysicalSortExpr], + properties: &EquivalenceProperties, + ) -> Result>> { + Ok(self.0.try_pushdown_sort(order, properties)?.map(Self::wrap)) + } + + fn reorder_files(&self, files: Vec) -> Vec { + self.0.reorder_files(files) + } + + fn try_pushdown_projection( + &self, + projection: &ProjectionExprs, + ) -> Result>> { + Ok(self.0.try_pushdown_projection(projection)?.map(Self::wrap)) + } + + fn apply_expressions( + &self, + f: &mut dyn FnMut(&Arc) -> Result, + ) -> Result { + self.0.apply_expressions(f) + } +} + +#[derive(Debug)] +struct FileContextMorselizer(Box); + +impl Morselizer for FileContextMorselizer { + fn plan_file(&self, file: PartitionedFile) -> Result> { + let path = file + .extensions + .get::() + .map(|path| Arc::clone(&path.0)) + .unwrap_or_else(|| Arc::from(file.object_meta.location.as_ref())); + Ok(Box::new(FileContext { + inner: self.0.plan_file(file)?, + path, + })) + } +} + +#[derive(Debug)] +struct FileContext { + inner: T, + path: Arc, +} + +impl MorselPlanner for FileContext> { + fn plan(self: Box) -> Result> { + let Self { inner, path } = *self; + let Some(mut plan) = inner.plan()? else { + return Ok(None); + }; + let morsels = plan + .take_morsels() + .into_iter() + .map(|inner| { + Box::new(FileContext { + inner, + path: Arc::clone(&path), + }) as Box + }) + .collect(); + let planners = plan + .take_ready_planners() + .into_iter() + .map(|inner| { + Box::new(FileContext { + inner, + path: Arc::clone(&path), + }) as Box + }) + .collect(); + if let Some(pending) = plan.take_pending_planner() { + plan.set_pending_planner(async move { + Ok(Box::new(FileContext { + inner: pending.await?, + path, + }) as Box) + }); + } + Ok(Some(plan.with_morsels(morsels).with_planners(planners))) + } +} + +impl Morsel for FileContext> { + fn into_stream(self: Box) -> BoxStream<'static, Result> { + let Self { inner, path } = *self; + inner + .into_stream() + .map_err(move |error| { + // Conversion runs inside this file's stream, before batches from other files mix. + if let DataFusionError::External(source) = &error { + if let Some(SparkError::ParquetTimestampOverflow { .. }) = + source.downcast_ref::() + { + return SparkError::ParquetTimestampOverflow { + file_path: path.to_string(), + } + .into(); + } + } + error + }) + .boxed() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[derive(Debug)] + struct FailedMorsel(SparkError); + + impl Morsel for FailedMorsel { + fn into_stream(self: Box) -> BoxStream<'static, Result> { + futures::stream::once(async move { Err(self.0.into()) }).boxed() + } + } + + #[tokio::test] + async fn file_paths_stay_with_their_streams() { + let stream = |path: &'static str, error| { + Box::new(FileContext { + inner: Box::new(FailedMorsel(error)) as Box, + path: Arc::from(path), + }) + .into_stream() + }; + let mut first = stream( + "file:///first%20file.parquet", + SparkError::ParquetTimestampOverflow { + file_path: String::new(), + }, + ); + let mut second = stream( + "s3a://bucket/second.parquet", + SparkError::ParquetTimestampOverflow { + file_path: String::new(), + }, + ); + // Consume in the opposite order from creation; no shared current-file state is involved. + for (result, path) in [ + (second.next().await, "s3a://bucket/second.parquet"), + (first.next().await, "file:///first%20file.parquet"), + ] { + let DataFusionError::External(error) = result.unwrap().unwrap_err() else { + panic!("expected a structured Spark error"); + }; + assert!(matches!(error.downcast_ref::(), + Some(SparkError::ParquetTimestampOverflow { file_path }) if file_path == path)); + } + let error = stream("file:///first%20file.parquet", SparkError::LongOverflow) + .next() + .await + .unwrap() + .unwrap_err(); + assert!(matches!(error, DataFusionError::External(source) + if matches!(source.downcast_ref::(), Some(SparkError::LongOverflow)))); + } +} diff --git a/native/core/src/parquet/mod.rs b/native/core/src/parquet/mod.rs index 7930320d148..4b4c2d2bd2f 100644 --- a/native/core/src/parquet/mod.rs +++ b/native/core/src/parquet/mod.rs @@ -24,5 +24,6 @@ pub mod schema_adapter; pub mod util; mod cast_column; +pub(crate) mod file_error_context; mod name_fold; pub(crate) mod objectstore; diff --git a/native/core/src/parquet/parquet_exec.rs b/native/core/src/parquet/parquet_exec.rs index 1ae9d47dd59..258a797c1d8 100644 --- a/native/core/src/parquet/parquet_exec.rs +++ b/native/core/src/parquet/parquet_exec.rs @@ -18,6 +18,7 @@ use crate::execution::operators::ExecutionError; use crate::parquet::eager_page_index_reader_factory::{EagerPageIndexReaderFactory, ScanIoSource}; use crate::parquet::encryption_support::{CometEncryptionConfig, ENCRYPTION_FACTORY_ID}; +use crate::parquet::file_error_context::ParquetErrorContext; use crate::parquet::name_fold::fold_schema_names; use crate::parquet::parquet_support::ObjectStoreBackend; use crate::parquet::parquet_support::SparkParquetOptions; @@ -221,6 +222,11 @@ pub(crate) fn init_datasource_exec( _ => Arc::new(parquet_source), }; + let file_source = if spark_parquet_options.checked_timestamp_overflow { + ParquetErrorContext::wrap(file_source) + } else { + file_source + }; let expr_adapter_factory: Arc = Arc::new( SparkPhysicalExprAdapterFactory::new(spark_parquet_options, default_values), ); diff --git a/native/core/src/parquet/parquet_support.rs b/native/core/src/parquet/parquet_support.rs index 1964f531746..66b214c2dee 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,11 @@ 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_else(|| SparkError::ParquetTimestampOverflow { + file_path: String::new(), + }) + })?; let micros = arrow::array::TimestampMicrosecondArray::new( micros.values().clone(), millis.nulls().cloned()) .with_timezone_opt(target_tz.clone()); @@ -1325,6 +1329,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 +1345,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 +1374,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/SparkErrorConverter.scala b/spark/src/main/scala/org/apache/comet/SparkErrorConverter.scala index a6bc21aca71..44010d42468 100644 --- a/spark/src/main/scala/org/apache/comet/SparkErrorConverter.scala +++ b/spark/src/main/scala/org/apache/comet/SparkErrorConverter.scala @@ -86,9 +86,7 @@ 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" && rawParams.get("filePath").forall(p => p == null || p.toString.isEmpty) @@ -117,8 +115,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..003ed152398 100644 --- a/spark/src/test/scala/org/apache/comet/SparkErrorConverterSuite.scala +++ b/spark/src/test/scala/org/apache/comet/SparkErrorConverterSuite.scala @@ -21,8 +21,41 @@ 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 path = "file:///tmp/data/bad%20timestamp%25.parquet" + val json = + s"""{"errorType":"ParquetTimestampOverflow","errorClass":"","params":{"filePath":"$path"}}""" + val ex = SparkErrorConverter.convertToSparkException( + new org.apache.comet.exceptions.CometQueryExecutionException(json), + taskFilePaths = Seq("file:///tmp/data/healthy.parquet", 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..a9ccdaa3a23 100644 --- a/spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala +++ b/spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala @@ -271,16 +271,14 @@ 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 => withTempDir { dir => - val path = new Path(dir.toURI.toString, "part-r-0.parquet") + val path = new Path(dir.toURI.toString, "bad timestamp%.parquet") + val healthyPath = new Path(dir.toURI.toString, "healthy.parquet") val schema = MessageTypeParser.parseMessageType(""" |message root { | optional int64 ts(TIMESTAMP_MILLIS); @@ -321,6 +319,13 @@ abstract class ParquetReadSuite extends CometTestBase { } writer.close() + val healthyWriter = createParquetWriter(schema, healthyPath, dictionaryEnabled) + val healthyRecord = new SimpleGroup(schema) + healthyRecord.add(0, 0L) + healthyRecord.add(1, 0L) + healthyWriter.write(healthyRecord) + healthyWriter.close() + val footerReader = org.apache.parquet.hadoop.ParquetFileReader.open( org.apache.parquet.hadoop.util.HadoopInputFile .fromPath(path, spark.sessionState.newHadoopConf())) @@ -338,16 +343,27 @@ 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) - assert(collect(selected.queryExecution.executedPlan) { - case _: CometNativeScanExec => true - }.nonEmpty) + withSQLConf( + SQLConf.ANSI_ENABLED.key -> ansiEnabled.toString, + SQLConf.FILES_MIN_PARTITION_NUM.key -> "1", + SQLConf.FILES_MAX_PARTITION_BYTES.key -> "134217728") { + readParquetFile(dir.toString) { df => + 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 => + val scans = collect(selected.queryExecution.executedPlan) { + case scan: CometNativeScanExec => scan + } + assert(scans.nonEmpty) + scans.foreach { scan => + assert(scan.perPartitionFilePaths.length == 1) + assert(scan.perPartitionFilePaths.head.size == 2) + } - val (sparkError, cometError) = checkSparkAnswerMaybeThrows(selected) - assert(Seq(sparkError, cometError).forall(_.exists(isOverflow))) + val error = checkSparkError(selected, errorClass) + assert(new java.net.URI(error.getMessageParameters.get("path")) == path.toUri) + assert(error.getCause.getClass == classOf[ArithmeticException]) + assert(error.getCause.getMessage == "long overflow") } } }