fix: validate Parquet VARIANT annotation against the requested read type

dev/diffs/4.1.3.diff

@@ -3231,29 +3231,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/join/tests.rs

@@ -667,6 +667,7 @@ fn parquet_probe(
         false,
         false,
         false,
+        false,
     )
     .unwrap();
     (file, scan)
@@ -772,6 +773,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

@@ -1737,6 +1737,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
@@ -222,7 +224,8 @@ pub(crate) fn init_datasource_exec(
     };
 
     let expr_adapter_factory: Arc<dyn PhysicalExprAdapterFactory> = Arc::new(
-        SparkPhysicalExprAdapterFactory::new(spark_parquet_options, default_values),
+        SparkPhysicalExprAdapterFactory::new(spark_parquet_options, default_values)
+            .with_required_schema(Arc::clone(&required_schema)),
     );
 
     let file_groups = file_groups
@@ -448,6 +451,7 @@ mod tests {
             false,
             false,
             false,
+            false,
         )
         .unwrap()
     }
@@ -622,6 +626,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()
@@ -349,3 +352,167 @@ async fn variant_scan_reads_wide_physical_decimal_as_decimal128() {
         }
     }
 }
+
+/// The two-field storage Spark writes for a Variant, declared without the Variant marker, which is
+/// the shape of a hand-written `struct<value binary, metadata binary>` read schema.
+fn plain_variant_storage() -> DataType {
+    DataType::Struct(Fields::from(vec![
+        Field::new("value", DataType::Binary, false),
+        Field::new("metadata", DataType::Binary, false),
+    ]))
+}
+
+/// The relation data schema `CometNativeScan` sends: every root, with `v` as a plain struct.
+fn id_and_plain_variant_schema() -> SchemaRef {
+    Arc::new(Schema::new(vec![
+        Field::new("id", DataType::Int32, true),
+        Field::new("v", plain_variant_storage(), true),
+    ]))
+}
+
+/// Writes `id INT` next to a VARIANT-annotated `v`. With `rows == 0` the file has no row group.
+fn write_id_and_annotated_variant(rows: usize) -> PathBuf {
+    let file_schema = Arc::new(Schema::new(vec![
+        Field::new("id", DataType::Int32, true),
+        Field::new("v", plain_variant_storage(), true).with_extension_type(VariantType),
+    ]));
+    let filename = get_temp_filename();
+    let mut writer = ArrowWriter::try_new_with_options(
+        File::create(&filename).unwrap(),
+        Arc::clone(&file_schema),
+        ArrowWriterOptions::new().with_skip_arrow_metadata(true),
+    )
+    .unwrap();
+    if rows > 0 {
+        let DataType::Struct(storage_fields) = plain_variant_storage() else {
+            unreachable!()
+        };
+        let variant = StructArray::new(
+            storage_fields,
+            vec![
+                Arc::new(BinaryArray::from(vec![Some(&[12u8, 1u8][..])])) as ArrayRef,
+                Arc::new(BinaryArray::from(vec![Some(&[1u8, 0u8, 0u8][..])])) as ArrayRef,
+            ],
+            None,
+        );
+        let batch = RecordBatch::try_new(
+            file_schema,
+            vec![
+                Arc::new(arrow::array::Int32Array::from(vec![1])) as ArrayRef,
+                Arc::new(variant) as ArrayRef,
+            ],
+        )
+        .unwrap();
+        writer.write(&batch).unwrap();
+    }
+    writer.close().unwrap();
+    filename
+}
+
+/// Scans through `init_datasource_exec` the way `CometNativeScan` drives it: the full data
+/// schema, the Spark read schema, and a projection into the data schema. Returns the row count or
+/// the error the scan raised.
+async fn scan_with_read_schema(
+    filename: PathBuf,
+    required_schema: SchemaRef,
+    projection: Vec<usize>,
+) -> Result<usize, datafusion::common::DataFusionError> {
+    let partitioned_file =
+        PartitionedFile::from_path(filename.to_string_lossy().into_owned()).unwrap();
+    let session_ctx = Arc::new(SessionContext::new());
+    let scan = init_datasource_exec(
+        required_schema,
+        Some(id_and_plain_variant_schema()),
+        None,
+        ObjectStoreUrl::local_filesystem(),
+        ObjectStoreBackend::Local,
+        vec![vec![partitioned_file]],
+        Some(projection),
+        None,
+        None,
+        "UTC",
+        false,
+        false,
+        false,
+        false,
+        &session_ctx,
+        false,
+        false,
+        false,
+        false,
+    )
+    .unwrap();
+    let mut stream = scan.execute(0, session_ctx.task_ctx())?;
+    let mut rows = 0;
+    while let Some(batch) = stream.next().await {
+        rows += batch?.num_rows();
+    }
+    Ok(rows)
+}
+
+fn read_schema_id_only() -> SchemaRef {
+    Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, true)]))
+}
+
+fn assert_variant_annotation_rejected(result: Result<usize, datafusion::common::DataFusionError>) {
+    let err = result.expect_err("reading the annotated `v` as a plain struct must be rejected");
+    assert!(
+        err.to_string().contains("_LEGACY_ERROR_TEMP_3071"),
+        "unexpected error: {err}"
+    );
+}
+
+/// Spark's schema pruning keeps unrequested roots in the relation data schema, so `v` reaches the
+/// native scan even when only `id` is read. The check must follow the read schema, not the data
+/// schema, or selecting `id` alone fails on a column the query never touches.
+#[tokio::test]
+async fn annotated_root_outside_the_read_schema_is_not_rejected() {
+    let rows = scan_with_read_schema(
+        write_id_and_annotated_variant(1),
+        read_schema_id_only(),
+        vec![0],
+    )
+    .await
+    .expect("`v` is not in the read schema, so its annotation must not be checked");
+    assert_eq!(rows, 1);
+}
+
+/// Once `v` is part of the read schema, the plain struct request is rejected as Spark rejects it.
+#[tokio::test]
+async fn annotated_root_inside_the_read_schema_is_rejected() {
+    assert_variant_annotation_rejected(
+        scan_with_read_schema(
+            write_id_and_annotated_variant(1),
+            id_and_plain_variant_schema(),
+            vec![0, 1],
+        )
+        .await,
+    );
+}
+
+/// Scoping the check to the read schema must not move it to row groups: a requested annotated root
+/// in a file with no row group is still rejected, matching Spark's schema-conversion-time check.
+#[tokio::test]
+async fn requested_annotated_root_is_rejected_on_an_empty_file() {
+    assert_variant_annotation_rejected(
+        scan_with_read_schema(
+            write_id_and_annotated_variant(0),
+            id_and_plain_variant_schema(),
+            vec![0, 1],
+        )
+        .await,
+    );
+}
+
+/// The empty-file counterpart of `annotated_root_outside_the_read_schema_is_not_rejected`.
+#[tokio::test]
+async fn unrequested_annotated_root_is_not_rejected_on_an_empty_file() {
+    let rows = scan_with_read_schema(
+        write_id_and_annotated_variant(0),
+        read_schema_id_only(),
+        vec![0],
+    )
+    .await
+    .expect("`v` is not in the read schema, so its annotation must not be checked");
+    assert_eq!(rows, 0);
+}