fix: reject a file without field ids at any depth whether or not id matching is on

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

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![],

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,13 +209,32 @@ impl EagerPageIndexReaderFactory {
             metadata_cache,
             scan_io_metrics,
             spark_variant_schema: false,
+            require_field_ids: false,
         }
     }
 
     pub fn with_spark_variant_schema(mut self, enabled: bool) -> Self {
         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<FileMetadataCache>,
     metadata_size_hint: Option<usize>,
     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<i32>| -> TypePtr {
+            Arc::new(
+                ParquetType::primitive_type_builder(name, PhysicalType::INT32)
+                    .with_id(id)
+                    .build()
+                    .unwrap(),
+            )
+        };
+        let group = |name: &str, id: Option<i32>, fields: Vec<TypePtr>| -> 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<dyn ObjectStore>,
+                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::<SparkError>(),
+                    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(

native/core/src/parquet/parquet_exec.rs

@@ -83,7 +83,7 @@ pub(crate) fn init_datasource_exec(
     session_ctx: &Arc<SessionContext>,
     encryption_enabled: bool,
     use_field_id: bool,
-    ignore_missing_field_id: bool,
+    require_field_ids: bool,
 ) -> Result<Arc<DataSourceExec>, ExecutionError> {
     // Computed once and reused below for `try_pushdown_filters`. `copied_config()` clones only
     // `SessionConfig` (an `Arc<ConfigOptions>` 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);