diff --git a/native/common/src/error.rs b/native/common/src/error.rs index acd7dac124d..8e6385065a3 100644 --- a/native/common/src/error.rs +++ b/native/common/src/error.rs @@ -233,11 +233,12 @@ pub enum SparkError { matched_fields: String, }, - /// The read schema requests Parquet field-id matching but the file carries no field ids. - /// Mirrors the runtime error raised in Spark's `ParquetReadSupport` when - /// `spark.sql.parquet.fieldId.read.ignoreMissing` is false. + /// The read schema carries Parquet field ids but the file carries none. Mirrors the runtime + /// error raised in Spark's `ParquetReadSupport` when + /// `spark.sql.parquet.fieldId.read.ignoreMissing` is false. Spark's message names no file, + /// so `file_path` travels as a parameter for the 4.x shim's `FAILED_READ_FILE` wrapper. #[error("Spark read schema expects field Ids, but Parquet file schema doesn't contain any field Ids. Please remove the field ids from Spark schema or ignore missing ids by setting `spark.sql.parquet.fieldId.read.ignoreMissing = true`")] - ParquetMissingFieldIds, + ParquetMissingFieldIds { file_path: String }, /// Schema mismatch when reading a Parquet column under a requested schema /// that's incompatible with the physical column type. Translated by the JVM @@ -353,7 +354,7 @@ impl SparkError { SparkError::FileNotFound { .. } => "FileNotFound", SparkError::DuplicateFieldCaseInsensitive { .. } => "DuplicateFieldCaseInsensitive", SparkError::DuplicateFieldByFieldId { .. } => "DuplicateFieldByFieldId", - SparkError::ParquetMissingFieldIds => "ParquetMissingFieldIds", + SparkError::ParquetMissingFieldIds { .. } => "ParquetMissingFieldIds", SparkError::ParquetSchemaConvert { .. } => "ParquetSchemaConvert", SparkError::CannotReadFile { .. } => "CannotReadFile", SparkError::Arrow(_) => "Arrow", @@ -604,6 +605,11 @@ impl SparkError { "matchedFields": matched_fields, }) } + SparkError::ParquetMissingFieldIds { file_path } => { + serde_json::json!({ + "filePath": file_path, + }) + } SparkError::ParquetSchemaConvert { file_path, column, @@ -711,10 +717,11 @@ impl SparkError { // (Spark's `foundDuplicateFieldInFieldIdLookupModeError` returns SparkRuntimeException) SparkError::DuplicateFieldByFieldId { .. } => "org/apache/spark/SparkRuntimeException", - // ParquetMissingFieldIds - converted to a plain RuntimeException by the shim, - // matching the `RuntimeException` Spark's ParquetReadSupport throws when the - // file lacks field ids and `spark.sql.parquet.fieldId.read.ignoreMissing=false`. - SparkError::ParquetMissingFieldIds => "java/lang/RuntimeException", + // ParquetMissingFieldIds - converted to the plain RuntimeException Spark's + // ParquetReadSupport throws when the file lacks field ids and + // `spark.sql.parquet.fieldId.read.ignoreMissing=false`. The 4.x shim wraps it in + // the FAILED_READ_FILE SparkException Spark 4 raises at the task boundary. + SparkError::ParquetMissingFieldIds { .. } => "java/lang/RuntimeException", // ParquetSchemaConvert - converted to SchemaColumnConvertNotSupportedException by the shim SparkError::ParquetSchemaConvert { .. } => { @@ -814,8 +821,9 @@ impl SparkError { // Duplicate field id in id-lookup mode SparkError::DuplicateFieldByFieldId { .. } => Some("_LEGACY_ERROR_TEMP_2094"), - // ParquetMissingFieldIds is a plain RuntimeException with no error class. - SparkError::ParquetMissingFieldIds => None, + // ParquetMissingFieldIds is a plain RuntimeException with no error class. The 4.x + // shim supplies FAILED_READ_FILE itself, so none is exposed here. + SparkError::ParquetMissingFieldIds { .. } => None, // Parquet schema mismatch — translated to SchemaColumnConvertNotSupportedException // by the JVM shim. The shim wraps it in the version-appropriate diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs index 10cc4ae37d7..b44420f47f4 100644 --- a/native/core/src/execution/planner.rs +++ b/native/core/src/execution/planner.rs @@ -1743,7 +1743,7 @@ impl PhysicalPlanner { self.session_ctx(), common.encryption_enabled, common.use_field_id, - common.ignore_missing_field_id, + common.require_field_ids, )?; Ok(( vec![], diff --git a/native/core/src/parquet/eager_page_index_reader_factory.rs b/native/core/src/parquet/eager_page_index_reader_factory.rs index d89a1772835..cadcf4c9f4e 100644 --- a/native/core/src/parquet/eager_page_index_reader_factory.rs +++ b/native/core/src/parquet/eager_page_index_reader_factory.rs @@ -43,8 +43,22 @@ //! fetch to encrypted scans that have no pruning predicate at all. Encrypted opens get exactly //! the caller's requested policy, unchanged from stock behavior. //! -//! Filed upstream as apache/datafusion#23978. Revert this once the opener merges its deferred -//! page-index load back into `FileMetadataCache` instead of bypassing it. +//! Filed upstream as apache/datafusion#23978. Once the opener merges its deferred page-index +//! load back into `FileMetadataCache` instead of bypassing it, the eager policy can go, but the +//! factory cannot: `get_metadata` is the one per-file hook that sees the raw footer, and two +//! other things hang off it. +//! +//! The first is Spark's missing field id check. `ParquetReadSupport` refuses to open a file +//! whose Parquet schema carries no field id when the requested schema carries one, unless +//! `ignoreMissing` is set, and it walks the raw `MessageType` to decide. That walk has to run +//! over the Parquet schema rather than the Arrow schema the schema adapter is handed later, +//! because an id on a repeated `list` or `key_value` group, or on the message root, never +//! reaches an Arrow field. Until apache/datafusion#24790, which is not in DataFusion 55.1.0, +//! the INT96 coercion also rebuilt container fields without their metadata, so a struct id could +//! vanish on the way to Arrow as well. +//! +//! The second is the Variant footer rewrite, `with_spark_arrow_schema`, which replaces the +//! Arrow schema hint in the footer for scans that project Variant. use arrow::datatypes::{DataType, FieldRef, Schema}; use async_trait::async_trait; @@ -58,6 +72,7 @@ use datafusion::execution::cache::cache_manager::FileMetadataCache; use datafusion::physical_plan::metrics::{ Count, ExecutionPlanMetricsSet, MetricBuilder, MetricCategory, MetricType, }; +use datafusion_comet_common::SparkError; use datafusion_datasource::PartitionedFile; use futures::future::BoxFuture; use futures::{FutureExt, StreamExt, TryStreamExt}; @@ -77,7 +92,7 @@ use parquet::file::metadata::{FileMetaData, KeyValue, ParquetMetaDataBuilder}; use parquet::file::metadata::{ FooterTail, PageIndexPolicy, ParquetMetaData, ParquetMetaDataReader, }; -use parquet::schema::types::{ColumnDescPtr, SchemaDescriptor}; +use parquet::schema::types::{ColumnDescPtr, SchemaDescriptor, Type as ParquetType}; use std::fmt::{Debug, Display, Formatter}; use std::ops::Range; use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; @@ -163,6 +178,9 @@ pub struct EagerPageIndexReaderFactory { // Enable the footer workaround only for scans that project Variant. // https://github.com/apache/datafusion-comet/issues/5477 spark_variant_schema: bool, + // Refuse a file whose Parquet schema carries no field id, as Spark's `ParquetReadSupport` + // does when the requested schema carries one and `ignoreMissing` is not set. + require_field_ids: bool, } impl EagerPageIndexReaderFactory { @@ -191,6 +209,7 @@ impl EagerPageIndexReaderFactory { metadata_cache, scan_io_metrics, spark_variant_schema: false, + require_field_ids: false, } } @@ -198,6 +217,24 @@ impl EagerPageIndexReaderFactory { self.spark_variant_schema = enabled; self } + + /// Refuse a file whose Parquet schema carries no field id. Off by default, so a factory + /// that never calls this reads every file. + pub fn with_require_field_ids(mut self, enabled: bool) -> Self { + self.require_field_ids = enabled; + self + } +} + +/// True when `node` or any node under it carries a field id, the way Spark's +/// `containsFieldIds` answers it over the raw Parquet schema, message root included. +fn contains_field_ids(node: &ParquetType) -> bool { + node.get_basic_info().has_id() + || (node.is_group() + && node + .get_fields() + .iter() + .any(|field| contains_field_ids(field))) } impl ParquetFileReaderFactory for EagerPageIndexReaderFactory { @@ -225,6 +262,7 @@ impl ParquetFileReaderFactory for EagerPageIndexReaderFactory { metadata_cache: Arc::clone(&self.metadata_cache), metadata_size_hint, spark_variant_schema: self.spark_variant_schema, + require_field_ids: self.require_field_ids, })) } } @@ -240,6 +278,7 @@ struct EagerPageIndexReader { metadata_cache: Arc, metadata_size_hint: Option, spark_variant_schema: bool, + require_field_ids: bool, } // Arrow infers ENUM as Binary, losing the distinction from raw binary that Spark needs. @@ -439,6 +478,7 @@ impl AsyncFileReader for EagerPageIndexReader { let metadata_size_hint = self.metadata_size_hint; let scan_io_metrics = Arc::clone(&self.scan_io_metrics); let spark_variant_schema = self.spark_variant_schema; + let require_field_ids = self.require_field_ids; async move { let file_decryption_properties = options .and_then(|o| o.file_decryption_properties()) @@ -498,6 +538,18 @@ impl AsyncFileReader for EagerPageIndexReader { } let metadata = metadata?; + // Spark's missing field id check, over the raw Parquet schema as in + // `ParquetReadSupport`. The JNI layer unwraps the `External` error, so the JVM sees + // the exception Spark raises. + if require_field_ids + && !contains_field_ids(metadata.file_metadata().schema_descr().root_schema()) + { + return Err(ParquetError::External(Box::new( + SparkError::ParquetMissingFieldIds { + file_path: object_meta.location.to_string(), + }, + ))); + } if spark_variant_schema { with_spark_arrow_schema(metadata) } else { @@ -833,7 +885,7 @@ mod tests { use arrow::{array::Int32Array, record_batch::RecordBatch}; use object_store::memory::InMemory; use parquet::{ - arrow::ArrowWriter, + arrow::{ArrowWriter, PARQUET_FIELD_ID_META_KEY}, file::{ properties::{EnabledStatistics, WriterProperties}, reader::FileReader, @@ -992,6 +1044,145 @@ mod tests { ); } + /// The raw schema walk answers like Spark's `containsFieldIds`: an id on the message root, + /// on a leaf, or on a repeated `list` or `key_value` group counts, and a schema without + /// any does not. + #[test] + fn contains_field_ids_sees_ids_on_any_node() { + use parquet::basic::Type as PhysicalType; + use parquet::schema::types::TypePtr; + + let leaf = |name: &str, id: Option| -> TypePtr { + Arc::new( + ParquetType::primitive_type_builder(name, PhysicalType::INT32) + .with_id(id) + .build() + .unwrap(), + ) + }; + let group = |name: &str, id: Option, fields: Vec| -> TypePtr { + Arc::new( + ParquetType::group_type_builder(name) + .with_id(id) + .with_fields(fields) + .build() + .unwrap(), + ) + }; + + assert!(!contains_field_ids(&group("schema", None, vec![]))); + assert!(!contains_field_ids(&group( + "schema", + None, + vec![leaf("a", None)] + ))); + assert!(contains_field_ids(&group( + "schema", + Some(1), + vec![leaf("a", None)] + ))); + assert!(contains_field_ids(&group( + "schema", + None, + vec![leaf("a", Some(1))] + ))); + let list_group_only = group( + "schema", + None, + vec![group( + "l", + None, + vec![group("list", Some(5), vec![leaf("element", None)])], + )], + ); + assert!(contains_field_ids(&list_group_only)); + let key_value_group_only = group( + "schema", + None, + vec![group( + "m", + None, + vec![group( + "key_value", + Some(6), + vec![leaf("key", None), leaf("value", None)], + )], + )], + ); + assert!(contains_field_ids(&key_value_group_only)); + let nested_without_ids = group( + "schema", + None, + vec![group("s", None, vec![leaf("a", None)])], + ); + assert!(!contains_field_ids(&nested_without_ids)); + } + + /// Write a one-column `a: int32` file into `store`, with a field id on `a` when `id` is + /// set, and return the file as `PartitionedFile`. + async fn put_int_file(store: &InMemory, location: &str, id: Option<&str>) -> PartitionedFile { + let mut field = arrow::datatypes::Field::new("a", DataType::Int32, false); + if let Some(id) = id { + field = field + .with_metadata([(PARQUET_FIELD_ID_META_KEY.to_string(), id.to_string())].into()); + } + let schema = Arc::new(Schema::new(vec![field])); + let batch = RecordBatch::try_new( + Arc::clone(&schema), + vec![Arc::new(Int32Array::from(vec![1, 2]))], + ) + .unwrap(); + let mut bytes = Vec::new(); + let mut writer = ArrowWriter::try_new(&mut bytes, schema, None).unwrap(); + writer.write(&batch).unwrap(); + writer.close().unwrap(); + let size = bytes.len() as u64; + store + .put(&Path::from(location), Bytes::from(bytes).into()) + .await + .unwrap(); + PartitionedFile::new(location.to_string(), size) + } + + /// A reader made by a factory with `require_field_ids` set refuses a file without ids on + /// its first metadata fetch, naming the file, and reads one that carries an id. A factory + /// without it reads the file without ids. + #[tokio::test] + async fn get_metadata_refuses_a_file_without_ids_only_when_required() { + let store = Arc::new(InMemory::new()); + let without_ids = put_int_file(&store, "no_ids.parquet", None).await; + let with_ids = put_int_file(&store, "ids.parquet", Some("1")).await; + let runtime = datafusion::execution::runtime_env::RuntimeEnv::default(); + let metrics = ExecutionPlanMetricsSet::new(); + let metadata_for = |require: bool, file: PartitionedFile| { + let factory = EagerPageIndexReaderFactory::new( + Arc::clone(&store) as Arc, + runtime.cache_manager.get_file_metadata_cache(), + ScanIoSource::Local, + &metrics, + ) + .with_require_field_ids(require); + let mut reader = factory.create_reader(0, file, None, &metrics).unwrap(); + async move { reader.get_metadata(None).await } + }; + + let err = metadata_for(true, without_ids.clone()) + .await + .expect_err("a file without ids must be refused"); + match err { + ParquetError::External(inner) => assert!( + matches!( + inner.downcast_ref::(), + Some(SparkError::ParquetMissingFieldIds { file_path }) if file_path == "no_ids.parquet" + ), + "unexpected error: {inner}" + ), + other => panic!("unexpected error: {other}"), + } + assert!(metadata_for(true, with_ids).await.is_ok()); + assert!(metadata_for(false, without_ids).await.is_ok()); + } + #[test] fn variant_policy_preserves_footer_metadata_and_indexes() { let schema = Arc::new(Schema::new_with_metadata( diff --git a/native/core/src/parquet/parquet_exec.rs b/native/core/src/parquet/parquet_exec.rs index 93ac29e824b..df621ccbc4a 100644 --- a/native/core/src/parquet/parquet_exec.rs +++ b/native/core/src/parquet/parquet_exec.rs @@ -83,7 +83,7 @@ pub(crate) fn init_datasource_exec( session_ctx: &Arc, encryption_enabled: bool, use_field_id: bool, - ignore_missing_field_id: bool, + require_field_ids: bool, ) -> Result, ExecutionError> { // Computed once and reused below for `try_pushdown_filters`. `copied_config()` clones only // `SessionConfig` (an `Arc` plus a small extensions map); `SessionContext:: @@ -101,7 +101,6 @@ pub(crate) fn init_datasource_exec( &session_config.options().execution.parquet, ); spark_parquet_options.use_field_id = use_field_id; - spark_parquet_options.ignore_missing_field_id = ignore_missing_field_id; // Spark can discard filtered-out values before timestamp conversion using statistics, // dictionary, and row-level filters. Comet cannot mirror every pruning path, so applying // checked conversion in a filtered scan can fail on values Spark never reads. Preserve the @@ -194,7 +193,8 @@ pub(crate) fn init_datasource_exec( scan_io_source, parquet_source.metrics(), ) - .with_spark_variant_schema(projects_variant), + .with_spark_variant_schema(projects_variant) + .with_require_field_ids(require_field_ids), ); parquet_source = parquet_source.with_parquet_file_reader_factory(reader_factory); diff --git a/native/core/src/parquet/parquet_support.rs b/native/core/src/parquet/parquet_support.rs index 89221c51ebb..e8987a79da6 100644 --- a/native/core/src/parquet/parquet_support.rs +++ b/native/core/src/parquet/parquet_support.rs @@ -104,10 +104,6 @@ pub struct SparkParquetOptions { /// (mirrors Spark's `spark.sql.parquet.fieldId.read.enabled`). Only takes effect /// when both physical and logical fields actually carry IDs. pub use_field_id: bool, - /// When false (Spark's default), reading a file that has no field ids while the - /// requested schema does carry ids raises a runtime error rather than silently - /// producing nulls (mirrors `spark.sql.parquet.fieldId.read.ignoreMissing`). - pub ignore_missing_field_id: bool, /// Whether type promotion (schema evolution) is allowed, e.g. INT32 -> INT64, /// FLOAT -> DOUBLE. Mirrors spark.comet.schemaEvolution.enabled. pub allow_type_promotion: bool, @@ -135,7 +131,6 @@ impl SparkParquetOptions { case_sensitive: false, return_null_struct_if_all_fields_missing: true, use_field_id: false, - ignore_missing_field_id: false, allow_type_promotion: false, allow_timestamp_ltz_to_ntz: false, checked_timestamp_overflow: true, @@ -152,7 +147,6 @@ impl SparkParquetOptions { case_sensitive: false, return_null_struct_if_all_fields_missing: true, use_field_id: false, - ignore_missing_field_id: false, allow_type_promotion: false, allow_timestamp_ltz_to_ntz: false, checked_timestamp_overflow: true, diff --git a/native/core/src/parquet/schema_adapter.rs b/native/core/src/parquet/schema_adapter.rs index 46d2ad7000d..c7ad0418578 100644 --- a/native/core/src/parquet/schema_adapter.rs +++ b/native/core/src/parquet/schema_adapter.rs @@ -69,7 +69,9 @@ impl SparkPhysicalExprAdapterFactory { } } -fn schema_has_field_ids(schema: &SchemaRef) -> bool { +/// True when a root field of `schema` carries a field id. Root only on purpose: it gates the +/// root name remap, and Spark's `clipParquetGroupFields` decides id matching one level at a time. +fn any_root_field_has_id(schema: &SchemaRef) -> bool { schema.fields().iter().any(|f| field_id(f).is_some()) } @@ -194,17 +196,8 @@ fn remap_physical_schema( physical_schema: &SchemaRef, case_sensitive: bool, use_field_id: bool, - ignore_missing_field_id: bool, ) -> DataFusionResult<(SchemaRef, HashMap)> { - let should_match_by_id = use_field_id && schema_has_field_ids(logical_schema); - - if should_match_by_id && !ignore_missing_field_id && !schema_has_field_ids(physical_schema) { - // Mirrors `ParquetReadSupport.inferSchema`'s eager check (Spark throws a runtime - // error rather than silently returning null columns). - return Err(DataFusionError::External(Box::new( - SparkError::ParquetMissingFieldIds, - ))); - } + let should_match_by_id = use_field_id && any_root_field_has_id(logical_schema); // Build id -> all matching physical field names. We need the full list so we can mirror // Spark's `_LEGACY_ERROR_TEMP_2094` "Found duplicate field(s)" error when an ID-bearing @@ -882,7 +875,7 @@ impl PhysicalExprAdapterFactory for SparkPhysicalExprAdapterFactory { // which uses the original physical file column names. let case_sensitive = self.parquet_options.case_sensitive; let should_match_by_id = - self.parquet_options.use_field_id && schema_has_field_ids(&logical_file_schema); + self.parquet_options.use_field_id && any_root_field_has_id(&logical_file_schema); let needs_remap = !case_sensitive || should_match_by_id; let (adapted_physical_schema, logical_to_physical_names) = if needs_remap { let (remapped, logical_to_physical) = remap_physical_schema( @@ -890,7 +883,6 @@ impl PhysicalExprAdapterFactory for SparkPhysicalExprAdapterFactory { &physical_file_schema, case_sensitive, self.parquet_options.use_field_id, - self.parquet_options.ignore_missing_field_id, )?; ( remapped, @@ -3473,7 +3465,7 @@ pub(crate) mod test { let logical = Arc::new(Schema::new(vec![Field::new("Name", DataType::Int32, true)])); let physical = Arc::new(Schema::new(vec![Field::new("NAME", DataType::Int32, true)])); let (remapped, name_map) = - super::remap_physical_schema(&logical, &physical, false, false, false).unwrap(); + super::remap_physical_schema(&logical, &physical, false, false).unwrap(); assert_eq!(remapped.field(0).name(), "Name"); assert_eq!(name_map.get("Name").map(String::as_str), Some("NAME")); } @@ -3490,7 +3482,7 @@ pub(crate) mod test { ])); let physical = Arc::new(Schema::new(vec![Field::new("FOO", DataType::Int32, true)])); let (remapped, _name_map) = - super::remap_physical_schema(&logical, &physical, false, true, true).unwrap(); + super::remap_physical_schema(&logical, &physical, false, true).unwrap(); assert!( remapped .field(0) @@ -3518,7 +3510,7 @@ pub(crate) mod test { Field::new("a", DataType::Int32, true).with_metadata(id_meta("9")) ])); let (remapped, _name_map) = - super::remap_physical_schema(&logical, &physical, true, true, false).unwrap(); + super::remap_physical_schema(&logical, &physical, true, true).unwrap(); assert_eq!(remapped.field(0).name(), "a"); } diff --git a/native/jni-bridge/src/errors.rs b/native/jni-bridge/src/errors.rs index 3e72f6e8048..199ad94246d 100644 --- a/native/jni-bridge/src/errors.rs +++ b/native/jni-bridge/src/errors.rs @@ -573,7 +573,9 @@ fn throw_exception(env: &mut Env, error: &CometError, backtrace: Option) // FAILED_READ_FILE / FileNotFound via the structured SparkError channel. Anything else // falls back to generic handling. CometError::DataFusion { msg: _, source } => { - if let Some(spark_error) = try_classify_file_read_error(source) { + if let Some(spark_error) = parquet_external_spark_error(source) { + throw_spark_error_as_json(env, spark_error) + } else if let Some(spark_error) = try_classify_file_read_error(source) { throw_spark_error_as_json(env, &spark_error) } else { throw_generic_exception(env, error, backtrace) @@ -646,6 +648,23 @@ fn throw_spark_error_as_json(env: &mut Env, spark_error: &SparkError) -> jni::er ) } +/// A `SparkError` the Parquet reader raised on open arrives as +/// `DataFusionError::ParquetError(ParquetError::External(spark_error))`. Unwrap it so the error +/// keeps its own JVM exception class instead of being classified as a file read failure. +/// `Context` and `Shared` wrappers are looked through, as `try_classify_file_read_error` does. +fn parquet_external_spark_error(error: &DataFusionError) -> Option<&SparkError> { + use datafusion::common::DataFusionError as DFE; + match error { + DFE::ParquetError(pe) => match pe.as_ref() { + ParquetError::External(inner) => inner.downcast_ref::(), + _ => None, + }, + DFE::Context(_, inner) => parquet_external_spark_error(inner), + DFE::Shared(inner) => parquet_external_spark_error(inner), + _ => None, + } +} + /// Classify a `DataFusionError` as a per-file read failure by TYPED variant (not message text), /// returning `SparkError::CannotReadFile` if so. This is the structured replacement for the /// previous JVM-side substring matching on error prose. @@ -1370,6 +1389,36 @@ mod tests { } } + /// A `SparkError` the Parquet reader raised on open stays typed through the `ParquetError` + /// wrapper and through `Context` and `Shared` wrappers, while an ordinary reader error does + /// not match. + #[test] + fn parquet_external_spark_error_keeps_its_type() { + let raised = DataFusionError::ParquetError(Box::new(ParquetError::External(Box::new( + SparkError::ParquetMissingFieldIds { + file_path: "a.parquet".to_string(), + }, + )))); + assert!(matches!( + parquet_external_spark_error(&raised), + Some(SparkError::ParquetMissingFieldIds { file_path }) if file_path == "a.parquet" + )); + let wrapped = DataFusionError::Context("open".to_string(), Box::new(raised)); + assert!(matches!( + parquet_external_spark_error(&wrapped), + Some(SparkError::ParquetMissingFieldIds { .. }) + )); + let shared = DataFusionError::Shared(Arc::new(wrapped)); + assert!(matches!( + parquet_external_spark_error(&shared), + Some(SparkError::ParquetMissingFieldIds { .. }) + )); + let corrupt = DataFusionError::ParquetError(Box::new(ParquetError::General( + "corrupt footer".to_string(), + ))); + assert!(parquet_external_spark_error(&corrupt).is_none()); + } + #[test] fn classify_parquet_error_is_file_read() { let e = DataFusionError::ParquetError(Box::new(parquet::errors::ParquetError::General( diff --git a/native/proto/src/proto/operator.proto b/native/proto/src/proto/operator.proto index 6a6284fb4c1..7b3a7aec0d2 100644 --- a/native/proto/src/proto/operator.proto +++ b/native/proto/src/proto/operator.proto @@ -163,8 +163,11 @@ message NativeScanCommon { // schema actually carries parquet.field.id metadata. When false the native // scan keeps its existing name-based path with no extra work. bool use_field_id = 15; - // True when spark.sql.parquet.fieldId.read.ignoreMissing is set. - bool ignore_missing_field_id = 16; + // True when the requested schema carries a field id and + // spark.sql.parquet.fieldId.read.ignoreMissing is not set. The native scan + // then refuses a file whose Parquet schema carries no field id, as Spark's + // ParquetReadSupport does whether or not the read flag is set. + bool require_field_ids = 16; // Whether widening type promotion is allowed (e.g. INT32 -> INT64, // FLOAT -> DOUBLE). Set from Comet's per-Spark-version constant in // ShimCometConf (false on 3.x, true on 4.x). When false, reading a column diff --git a/spark/src/main/scala/org/apache/comet/serde/operator/CometNativeScan.scala b/spark/src/main/scala/org/apache/comet/serde/operator/CometNativeScan.scala index de8652ad90c..080c86085b3 100644 --- a/spark/src/main/scala/org/apache/comet/serde/operator/CometNativeScan.scala +++ b/spark/src/main/scala/org/apache/comet/serde/operator/CometNativeScan.scala @@ -294,12 +294,13 @@ object CometNativeScan extends CometOperatorSerde[CometScanExec] with CometTypeS // Field-ID matching: only ask the native side to do extra work when the conf is on AND // the requested schema actually carries IDs. Spark's ParquetReadSupport applies the same // gate before invoking matchIdField. - val useFieldId = - scan.conf.getConf(SQLConf.PARQUET_FIELD_ID_READ_ENABLED) && - ParquetUtils.hasFieldIds(scan.requiredSchema) + val hasFieldIds = ParquetUtils.hasFieldIds(scan.requiredSchema) + val useFieldId = scan.conf.getConf(SQLConf.PARQUET_FIELD_ID_READ_ENABLED) && hasFieldIds commonBuilder.setUseFieldId(useFieldId) - commonBuilder.setIgnoreMissingFieldId( - scan.conf.getConf(SQLConf.IGNORE_MISSING_PARQUET_FIELD_ID)) + // Spark's ParquetReadSupport refuses a file without field ids whenever the requested + // schema carries one, whatever the read flag says, unless ignoreMissing is set. + commonBuilder.setRequireFieldIds( + hasFieldIds && !scan.conf.getConf(SQLConf.IGNORE_MISSING_PARQUET_FIELD_ID)) commonBuilder.setAllowTypePromotion(CometConf.COMET_SCHEMA_EVOLUTION_ENABLED) commonBuilder.setAllowTimestampLtzToNtz(CometConf.COMET_ALLOW_TIMESTAMP_LTZ_AS_NTZ) 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 287d4ecb14a..c5b1d04f540 100644 --- a/spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala +++ b/spark/src/test/scala/org/apache/comet/parquet/ParquetReadSuite.scala @@ -21,6 +21,7 @@ package org.apache.comet.parquet import java.io.File import java.math.{BigDecimal, BigInteger} +import java.sql.Timestamp import java.time.{ZoneId, ZoneOffset} import java.util.{Base64, Collections} @@ -2251,45 +2252,229 @@ abstract class ParquetReadSuite extends CometTestBase { } } - // Verbatim port of Spark `ParquetFieldIdIOSuite.test("read parquet file without ids")`, - // for the same reason as the duplicate-id test above. + // Port of Spark `ParquetFieldIdIOSuite.test("read parquet file without ids")`, for the same + // reason as the duplicate-id test above. It runs with the read flag off as well, since Spark's + // `ParquetReadSupport` checks for missing ids before it consults the flag, and with id zero as + // one of the read schemas, since zero is an id like any other. test("read parquet file without ids") { + Seq("true", "false").foreach { readEnabled => + withSQLConf(SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> readEnabled) { + withTempPath { dir => + val readSchema = + new StructType() + .add("a", IntegerType, true, withId(1)) + + val writeSchema = + new StructType() + .add("a", IntegerType, true) + .add("rand1", StringType, true) + .add("rand2", StringType, true) + + val writeData = Seq(Row(100, "text", "txt"), Row(200, "more", "mr")) + spark + .createDataFrame(spark.sparkContext.parallelize(writeData), writeSchema) + .write + .mode("overwrite") + .parquet(dir.getCanonicalPath) + + val idZeroSchema = new StructType().add("a", IntegerType, true, withId(0)) + Seq(readSchema, readSchema.add("b", StringType, true), idZeroSchema).foreach { schema => + withClue(s"read flag $readEnabled, schema $schema: ") { + val cause = intercept[SparkException] { + spark.read.schema(schema).parquet(dir.getCanonicalPath).collect() + }.getCause + assert(cause.isInstanceOf[RuntimeException] && + cause.getMessage.contains("Parquet file schema doesn't contain any field Ids")) + withSQLConf(SQLConf.IGNORE_MISSING_PARQUET_FIELD_ID.key -> "true") { + val df = spark.read.schema(schema).parquet(dir.getCanonicalPath) + if (readEnabled.toBoolean) { + // Spark's own assertion: no file field carries a requested id, so every + // column is null filled. + val expectedValues = (1 to schema.length).map(_ => null) + checkAnswer(df, Row(expectedValues: _*) :: Row(expectedValues: _*) :: Nil) + } else { + checkSparkAnswerAndOperator(df) + } + } + } + } + } + } + } + } + + // Spark's `containsFieldIds` walks the whole file schema, so ids that sit only on struct + // children count. The root field whose id the file lacks is null filled rather than rejected. + test("a file whose field ids are only on nested fields reads without a missing-id error") { withSQLConf(SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> "true") { withTempPath { dir => - val readSchema = - new StructType() - .add("a", IntegerType, true, withId(1)) + val nested = StructType(Seq(StructField("a", IntegerType, nullable = true, withId(11)))) + val writeSchema = new StructType().add("s", nested, true) + val readSchema = new StructType() + .add("s", nested, true) + .add("missing", IntegerType, true, withId(7)) + val writeData = Seq(Row(Row(1)), Row(Row(2))) + spark + .createDataFrame(spark.sparkContext.parallelize(writeData), writeSchema) + .write + .mode("overwrite") + .parquet(dir.getCanonicalPath) - val writeSchema = - new StructType() - .add("a", IntegerType, true) - .add("rand1", StringType, true) - .add("rand2", StringType, true) + checkSparkAnswerAndOperator(spark.read.schema(readSchema).parquet(dir.getCanonicalPath)) + } + } + } - val writeData = Seq(Row(100, "text", "txt"), Row(200, "more", "mr")) + // Spark's `containsFieldIds` walks the raw Parquet schema, where an id may sit on the + // repeated `list` or `key_value` group of a list or map, which no Spark or Arrow field ever + // shows. Such a file carries ids, so it is not rejected. With the read flag on, the root + // fields ask for ids that no root field of the file carries and are null filled. + test("ids on repeated list and key_value groups count as file ids") { + withTempDir { dir => + val path = new Path(dir.toURI.toString, "part-r-0.parquet") + val schema = MessageTypeParser.parseMessageType(""" + |message schema { + | optional group l (LIST) { + | repeated group list = 5 { + | optional int32 element; + | } + | } + | optional group m (MAP) { + | repeated group key_value = 6 { + | required int32 key; + | optional int32 value; + | } + | } + |} + |""".stripMargin) + val writer = createParquetWriter(schema, path) + (1 to 2).foreach { i => + val record = new SimpleGroup(schema) + record.addGroup(0).addGroup(0).add(0, i) + val entry = record.addGroup(1).addGroup(0) + entry.add(0, i) + entry.add(1, i * 10) + writer.write(record) + } + writer.close() + + val readSchema = new StructType() + .add("l", ArrayType(IntegerType), true, withId(5)) + .add("m", MapType(IntegerType, IntegerType), true, withId(6)) + Seq("false", "true").foreach { readEnabled => + withSQLConf(SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> readEnabled) { + withClue(s"read flag $readEnabled: ") { + checkSparkAnswerAndOperator( + spark.read.schema(readSchema).parquet(dir.getCanonicalPath)) + } + } + } + } + } + + // Spark checks for missing ids against the pruned schema it hands the reader, so an id on a + // column or struct child the query never reads does not reject a file without ids, and a + // count reads no column at all. A read that touches an id-bearing field still raises. + test("missing ids are checked against the pruned read schema") { + withSQLConf(SQLConf.NESTED_SCHEMA_PRUNING_ENABLED.key -> "true") { + withTempPath { dir => + val writeSchema = new StructType() + .add("a", IntegerType) + .add("b", IntegerType) + .add("s", new StructType().add("a", IntegerType).add("b", IntegerType)) + val readSchema = new StructType() + .add("a", IntegerType, true, withId(1)) + .add("b", IntegerType) + .add( + "s", + new StructType().add("a", IntegerType, true, withId(11)).add("b", IntegerType)) + val writeData = Seq(Row(1, 2, Row(3, 4)), Row(5, 6, Row(7, 8))) spark .createDataFrame(spark.sparkContext.parallelize(writeData), writeSchema) .write .mode("overwrite") .parquet(dir.getCanonicalPath) - Seq(readSchema, readSchema.add("b", StringType, true)).foreach { schema => - val cause = intercept[SparkException] { - spark.read.schema(schema).parquet(dir.getCanonicalPath).collect() - }.getCause - assert( - cause.isInstanceOf[RuntimeException] && - cause.getMessage.contains("Parquet file schema doesn't contain any field Ids")) - val expectedValues = (1 to schema.length).map(_ => null) - withSQLConf(SQLConf.IGNORE_MISSING_PARQUET_FIELD_ID.key -> "true") { - checkAnswer( - spark.read.schema(schema).parquet(dir.getCanonicalPath), - Row(expectedValues: _*) :: Row(expectedValues: _*) :: Nil) - } + def read(): DataFrame = spark.read.schema(readSchema).parquet(dir.getCanonicalPath) + checkSparkAnswerAndOperator(read().select("b")) + checkSparkAnswerAndOperator(read().select("s.b")) + checkSparkAnswerAndOperator(read().selectExpr("count(*)")) + assertMissingFieldIds(read().select("a")) + assertMissingFieldIds(read().select("s.a")) + } + } + } + + // Spark writes timestamps as INT96 by default. The id on `s` is in the Parquet schema, which + // the check reads, so the file opens and `s` resolves by name. + test("field ids on a struct holding a timestamp survive the missing-id check") { + withSQLConf(SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> "false") { + withTempPath { dir => + val nested = new StructType().add("a", IntegerType).add("ts", TimestampType) + val schema = new StructType().add("s", nested, true, withId(1)) + val ts = Timestamp.valueOf("2020-01-01 00:00:00") + val writeData = Seq(Row(Row(1, ts)), Row(Row(2, ts))) + spark + .createDataFrame(spark.sparkContext.parallelize(writeData), schema) + .write + .mode("overwrite") + .parquet(dir.getCanonicalPath) + + checkSparkAnswerAndOperator(spark.read.schema(schema).parquet(dir.getCanonicalPath)) + } + } + } + + // Spark checks each file on its own. A directory holding one file with ids and one without + // raises on the second, and with `ignoreMissing` the file without ids reads as nulls because + // no root field of it carries the requested id. + test("a file without ids next to a file with ids is checked on its own") { + withSQLConf(SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key -> "true") { + withTempPath { dir => + val idSchema = new StructType().add("x", IntegerType, true, withId(1)) + val plainSchema = new StructType().add("a", IntegerType, true) + val readSchema = new StructType().add("a", IntegerType, true, withId(1)) + spark + .createDataFrame(spark.sparkContext.parallelize(Seq(Row(100), Row(200))), idSchema) + .write + .mode("overwrite") + .parquet(dir.getCanonicalPath) + spark + .createDataFrame(spark.sparkContext.parallelize(Seq(Row(1), Row(2))), plainSchema) + .write + .mode("append") + .parquet(dir.getCanonicalPath) + + assertMissingFieldIds(spark.read.schema(readSchema).parquet(dir.getCanonicalPath)) + + withSQLConf(SQLConf.IGNORE_MISSING_PARQUET_FIELD_ID.key -> "true") { + val df = spark.read.schema(readSchema).parquet(dir.getCanonicalPath) + checkSparkAnswerAndOperator(df) + checkAnswer(df, Row(100) :: Row(200) :: Row(null) :: Row(null) :: Nil) } } } } + + // Spark's own assertion from `ParquetFieldIdIOSuite`: the `SparkException` a read raises has + // the `RuntimeException` from `ParquetReadSupport` as its cause. + private def isMissingFieldIdsError(error: Throwable): Boolean = { + val cause = error.getCause + cause.isInstanceOf[RuntimeException] && + cause.getMessage.contains("Parquet file schema doesn't contain any field Ids") + } + + // Spark and Comet both raise the missing field ids error for `df`, and Comet plans the read + // natively, so the error comes from the native check rather than from a fallback to Spark. + private def assertMissingFieldIds(df: => DataFrame): Unit = { + checkCometOperators(stripAQEPlan(df.queryExecution.executedPlan)) + val (sparkError, cometError) = checkSparkAnswerMaybeThrows(df) + Seq("Spark" -> sparkError, "Comet" -> cometError).foreach { case (engine, error) => + assert( + error.exists(isMissingFieldIdsError), + s"$engine: " + error.map(causeChain(_).mkString("\n ")).getOrElse("no error")) + } + } } class ParquetReadV1Suite extends ParquetReadSuite with AdaptiveSparkPlanHelper {