fix: validate Parquet VARIANT annotation against the requested read type

dev/diffs/4.1.3.diff

@@ -3212,29 +3212,6 @@ index 09ed6955a51..52d998aab46 100644
      )
    }
    test(s"parquet widening conversion $fromType -> $toType") {
-diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVariantShreddingSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVariantShreddingSuite.scala
-index 1cc6d3afbee..1ca791bc0cc 100644
---- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVariantShreddingSuite.scala
-+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetVariantShreddingSuite.scala
-@@ -29,7 +29,7 @@ import org.apache.parquet.schema.{LogicalTypeAnnotation, PrimitiveType, Type}
- import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName
-
- import org.apache.spark.SparkException
--import org.apache.spark.sql.{AnalysisException, QueryTest, Row}
-+import org.apache.spark.sql.{AnalysisException, IgnoreComet, QueryTest, Row}
- import org.apache.spark.sql.internal.SQLConf
- import org.apache.spark.sql.internal.SQLConf.ParquetOutputTimestampType
- import org.apache.spark.sql.test.SharedSparkSession
-@@ -230,7 +230,8 @@ class ParquetVariantShreddingSuite extends QueryTest with ParquetTest with Share
-     }
-   }
-
--  test("variant logical type annotation - ignore variant annotation") {
-+  test("variant logical type annotation - ignore variant annotation",
-+    IgnoreComet("https://github.com/apache/datafusion-comet/issues/5741")) {
-     Seq(true, false).foreach { ignoreVariantAnnotation =>
-       withSQLConf(SQLConf.PARQUET_ANNOTATE_VARIANT_LOGICAL_TYPE.key -> "true",
-         SQLConf.PARQUET_IGNORE_VARIANT_ANNOTATION.key -> ignoreVariantAnnotation.toString,
 diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/debug/DebuggingSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/debug/DebuggingSuite.scala
 index b8f3ea3c6f3..bbd44221288 100644
 --- a/sql/core/src/test/scala/org/apache/spark/sql/execution/debug/DebuggingSuite.scala

native/common/src/error.rs

@@ -234,6 +234,18 @@ pub enum SparkError {
     #[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,
 
+    /// A Parquet field carries the VARIANT logical type annotation but the requested Spark read
+    /// type is not `VariantType`, while `spark.sql.parquet.ignoreVariantAnnotation` is false.
+    /// Mirrors the `checkConversionRequirement` in Spark's
+    /// `ParquetToSparkSchemaConverter.convertGroupField`, which raises `_LEGACY_ERROR_TEMP_3071`
+    /// during schema conversion -- before any row is read, so an empty file is rejected too.
+    #[error("[_LEGACY_ERROR_TEMP_3071] Invalid Spark read type: expected {column} to be variant type but found {spark_type}")]
+    ParquetVariantAnnotationMismatch {
+        file_path: String,
+        column: String,
+        spark_type: String,
+    },
+
     /// Schema mismatch when reading a Parquet column under a requested schema
     /// that's incompatible with the physical column type. Translated by the JVM
     /// shim into Spark's `SchemaColumnConvertNotSupportedException`. The
@@ -348,6 +360,9 @@ impl SparkError {
             SparkError::DuplicateFieldCaseInsensitive { .. } => "DuplicateFieldCaseInsensitive",
             SparkError::DuplicateFieldByFieldId { .. } => "DuplicateFieldByFieldId",
             SparkError::ParquetMissingFieldIds => "ParquetMissingFieldIds",
+            SparkError::ParquetVariantAnnotationMismatch { .. } => {
+                "ParquetVariantAnnotationMismatch"
+            }
             SparkError::ParquetSchemaConvert { .. } => "ParquetSchemaConvert",
             SparkError::CannotReadFile { .. } => "CannotReadFile",
             SparkError::Arrow(_) => "Arrow",
@@ -598,6 +613,17 @@ impl SparkError {
                     "matchedFields": matched_fields,
                 })
             }
+            SparkError::ParquetVariantAnnotationMismatch {
+                file_path,
+                column,
+                spark_type,
+            } => {
+                serde_json::json!({
+                    "filePath": file_path,
+                    "column": column,
+                    "sparkType": spark_type,
+                })
+            }
             SparkError::ParquetSchemaConvert {
                 file_path,
                 column,
@@ -709,6 +735,13 @@ impl SparkError {
             // file lacks field ids and `spark.sql.parquet.fieldId.read.ignoreMissing=false`.
             SparkError::ParquetMissingFieldIds => "java/lang/RuntimeException",
 
+            // ParquetVariantAnnotationMismatch - the shim rebuilds Spark's AnalysisException and
+            // wraps it in a FAILED_READ_FILE SparkException, matching what Spark's own file scan
+            // produces when its schema converter rejects the read type.
+            SparkError::ParquetVariantAnnotationMismatch { .. } => {
+                "org/apache/spark/sql/AnalysisException"
+            }
+
             // ParquetSchemaConvert - converted to SchemaColumnConvertNotSupportedException by the shim
             SparkError::ParquetSchemaConvert { .. } => {
                 "org/apache/spark/sql/execution/datasources/SchemaColumnConvertNotSupportedException"
@@ -812,6 +845,10 @@ impl SparkError {
             // Parquet schema mismatch — translated to SchemaColumnConvertNotSupportedException
             // by the JVM shim. The shim wraps it in the version-appropriate
             // SparkException error class, so no error class is exposed here.
+            // ParquetVariantAnnotationMismatch — the shim rebuilds Spark's AnalysisException with
+            // its own error class and wraps it via cannotReadFilesError, so none is exposed here.
+            SparkError::ParquetVariantAnnotationMismatch { .. } => None,
+
             SparkError::ParquetSchemaConvert { .. } => None,
 
             // CannotReadFile — the JVM shim wraps it via cannotReadFilesError, which supplies the

native/core/src/execution/operators/dynamic_filter/tests.rs

@@ -701,6 +701,7 @@ fn parquet_probe(
         false,
         false,
         false,
+        false,
     )
     .unwrap();
     (file, scan)
@@ -858,6 +859,7 @@ async fn reader_filter_crosses_null_check_conjunction_and_retains_residual() {
             false,
             false,
             false,
+            false,
         )
         .unwrap();
         let checks = [("key", 0), ("payload", 1), ("other", 2)].map(|(name, index)| {

native/core/src/execution/operators/iceberg_scan.rs

@@ -230,6 +230,14 @@ impl IcebergScanExec {
         let scan_metrics = scan_result.metrics().clone();
         let stream = scan_result.stream();
 
+        // `ignore_variant_annotation` stays at its default of false, so the Parquet VARIANT
+        // annotation check in `SparkPhysicalExprAdapterFactory::create` is always enforced here.
+        // `spark.sql.parquet.ignoreVariantAnnotation` is a Parquet-datasource conf that the
+        // Iceberg read path never consults, so there is deliberately no escape hatch: an Iceberg
+        // table whose file annotates a field as VARIANT while the table schema asks for an
+        // ordinary struct is the same defect apache/datafusion-comet#5741 describes. Reaching it
+        // needs a table schema that disagrees with its own data files, since `CometScanRule` falls
+        // back for any Iceberg table carrying a Variant column.
         let spark_options = SparkParquetOptions::new(EvalMode::Legacy, "UTC", false);
         let adapter_factory = SparkPhysicalExprAdapterFactory::new(spark_options, None);
 

native/core/src/execution/planner.rs

@@ -1734,6 +1734,7 @@ impl PhysicalPlanner {
                     common.encryption_enabled,
                     common.use_field_id,
                     common.ignore_missing_field_id,
+                    common.ignore_variant_annotation,
                 )?;
                 Ok((
                     vec![],

native/core/src/parquet/parquet_exec.rs

@@ -83,6 +83,7 @@ pub(crate) fn init_datasource_exec(
     encryption_enabled: bool,
     use_field_id: bool,
     ignore_missing_field_id: bool,
+    ignore_variant_annotation: 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,6 +102,7 @@ pub(crate) fn init_datasource_exec(
     );
     spark_parquet_options.use_field_id = use_field_id;
     spark_parquet_options.ignore_missing_field_id = ignore_missing_field_id;
+    spark_parquet_options.ignore_variant_annotation = ignore_variant_annotation;
     // 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
@@ -448,6 +450,7 @@ mod tests {
             false,
             false,
             false,
+            false,
         )
         .unwrap()
     }
@@ -622,6 +625,7 @@ mod tests {
             false,
             false,
             false,
+            false,
         )
         .unwrap();
 

native/core/src/parquet/parquet_exec/variant_tests.rs

@@ -163,6 +163,7 @@ async fn scan_variant_file(filename: PathBuf) -> VariantArray {
         false,
         false,
         false,
+        false,
     )
     .unwrap();
     let mut stream = scan.execute(0, session_ctx.task_ctx()).unwrap();
@@ -228,6 +229,7 @@ async fn unread_variant_does_not_override_arrow_schema_hint() {
         false,
         false,
         false,
+        false,
     )
     .unwrap();
     let mut stream = scan.execute(0, session.task_ctx()).unwrap();
@@ -258,6 +260,7 @@ fn encrypted_projected_variant_is_rejected_before_reader_creation() {
         true,
         false,
         false,
+        false,
     );
     assert!(result
         .unwrap_err()

native/core/src/parquet/parquet_support.rs

@@ -112,6 +112,11 @@ pub struct SparkParquetOptions {
     /// (overflow -> NULL), because Spark may discard values through pruning paths that
     /// DataFusion cannot fully mirror before conversion.
     pub checked_timestamp_overflow: bool,
+    /// When true (`spark.sql.parquet.ignoreVariantAnnotation`, Spark 4.1+), a Parquet field
+    /// carrying the VARIANT logical type annotation is read as its plain underlying struct.
+    /// When false (Spark's default), requesting anything but Spark's VariantType for such a
+    /// field is rejected, mirroring `ParquetToSparkSchemaConverter.convertGroupField`.
+    pub ignore_variant_annotation: bool,
 }
 
 impl SparkParquetOptions {
@@ -129,6 +134,7 @@ impl SparkParquetOptions {
             allow_type_promotion: false,
             allow_timestamp_ltz_to_ntz: false,
             checked_timestamp_overflow: true,
+            ignore_variant_annotation: false,
         }
     }
 
@@ -146,6 +152,7 @@ impl SparkParquetOptions {
             allow_type_promotion: false,
             allow_timestamp_ltz_to_ntz: false,
             checked_timestamp_overflow: true,
+            ignore_variant_annotation: false,
         }
     }
 }