From 94641069685f1bc044586ccc0c36cf402a960d23 Mon Sep 17 00:00:00 2001 From: dustin Date: Sat, 26 Sep 2026 21:38:18 +0000 Subject: [PATCH] fix: reject a file without field ids at any depth whether or not id matching is on (#6116) * fix: reject a file without field ids at any depth whether or not id matching is on Spark's `ParquetReadSupport.getRequestedSchema` raises when the requested schema carries a Parquet field id at any depth and the file carries none at any depth, unless `spark.sql.parquet.fieldId.read.ignoreMissing` is set. The check runs on every read, for both readers, and does not consult `spark.sql.parquet.fieldId.read.enabled`. The native scan ran that check only with the read flag on, only over root fields, and only inside the name remap, which a case-sensitive session with the flag off never reaches. A read schema with ids over a file without ids returned rows where Spark raises, and a file whose ids sit only on nested fields was rejected where Spark reads it and null-fills the unmatched fields. The check now runs at the top of the expression adapter factory with a recursive predicate on both schemas. Id matching itself stays gated on the read flag and root ids, as Spark gates it per struct level. The Iceberg scan opts out, since its reader resolves columns by id and supplies the ids itself. * fix: answer both halves of the missing-field-id check from what Spark reads Spark checks the pruned required schema for ids and the raw Parquet schema for their absence. The requested half now comes from `required_schema` at plan time, since the logical schema DataFusion hands the adapter is the full read schema, so an id on a column the query never projects no longer raises. The file half moves into the eager page index reader, which walks the raw schema from the footer. The Arrow schema the adapter sees has lost container metadata after INT96 coercion and never carries ids on repeated list or key value groups, so a file whose only id sat on a struct holding a timestamp raised where Spark reads. The JNI error conversion unwraps the Parquet external error so the Java side keeps the same exception, and the Iceberg scan no longer needs an opt-out since it does not use that reader. Tests cover the pruned projection, the timestamp struct, an id only on a list or key value group, a directory mixing files with and without ids, nested schema pruning, id zero, count(*), and the exception class on each version. * test: find the missing field ids error anywhere in the cause chain On Spark 4.0 and 4.1 the RuntimeException Spark raises for a read schema with field ids over a file without any arrives wrapped once, in the FAILED_READ_FILE SparkException that collect raises, and the helper assumed one more layer above it. It now walks the cause chain and requires a plain RuntimeException carrying Spark's message, for Spark and for Comet alike, so the layer count no longer matters. * fix: let the serde say whether the file must carry field ids The requested side of the missing field ids check is Spark's own hasFieldIds over the required schema, so the serde now sends one flag, require_field_ids, in place of ignore_missing_field_id, and the native Arrow walk is gone. The reader factory takes that one flag through a builder and the parquet options carry nothing for it. The error names the file, and the module doc leads with the lasting reason the check reads the raw footer: ids on repeated groups and on the message root never reach Arrow. The Rust scan tests that had Scala twins are gone, the repeated group cases moved to Scala where they compare against Spark, and the Scala tests fold into the existing port and one pruned schema test with Spark's own assertion form. (cherry picked from commit 863f11de5eb78bacf11737032cdbcf496ad520b5) --- native/common/src/error.rs | 30 ++- native/core/src/execution/planner.rs | 2 +- .../eager_page_index_reader_factory.rs | 199 ++++++++++++++- native/core/src/parquet/parquet_exec.rs | 6 +- native/core/src/parquet/parquet_support.rs | 6 - native/core/src/parquet/schema_adapter.rs | 24 +- native/jni-bridge/src/errors.rs | 51 +++- native/proto/src/proto/operator.proto | 7 +- .../serde/operator/CometNativeScan.scala | 11 +- .../comet/parquet/ParquetReadSuite.scala | 233 ++++++++++++++++-- 10 files changed, 496 insertions(+), 73 deletions(-) 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 {