feat: support direct Variant projection in native Parquet scans

.github/workflows/pr_build_linux.yml

@@ -373,6 +373,7 @@ jobs:
               org.apache.spark.sql.comet.ParquetDatetimeRebaseV2Suite
               org.apache.spark.sql.comet.ParquetEncryptionITCase
               org.apache.comet.exec.CometNativeReaderSuite
+              org.apache.comet.CometVariantProjectionSuite
               org.apache.comet.CometIcebergNativeSuite
               org.apache.comet.CometIcebergEncryptionSuite
               org.apache.comet.CometIcebergRewriteActionSuite

.github/workflows/pr_build_macos.yml

@@ -127,6 +127,7 @@ jobs:
               org.apache.spark.sql.comet.ParquetDatetimeRebaseV2Suite
               org.apache.spark.sql.comet.ParquetEncryptionITCase
               org.apache.comet.exec.CometNativeReaderSuite
+              org.apache.comet.CometVariantProjectionSuite
               org.apache.comet.CometIcebergNativeSuite
               org.apache.comet.CometIcebergEncryptionSuite
               org.apache.comet.CometIcebergRewriteActionSuite

dev/diffs/4.1.3.diff

@@ -1208,6 +1208,20 @@ index e4b5e10f7c3..c6efde09c8a 100644
 
    protected val baseResourcePath = {
      // use the same way as `SQLQueryTestSuite` to get the resource path
+diff --git a/sql/core/src/test/scala/org/apache/spark/sql/ResolveDefaultColumnsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/ResolveDefaultColumnsSuite.scala
+index cb9d0909554..084d6515e8b 100644
+--- a/sql/core/src/test/scala/org/apache/spark/sql/ResolveDefaultColumnsSuite.scala
++++ b/sql/core/src/test/scala/org/apache/spark/sql/ResolveDefaultColumnsSuite.scala
+@@ -284,7 +284,8 @@ class ResolveDefaultColumnsSuite extends QueryTest with SharedSparkSession {
+     withTable("t") {
+       sql("CREATE TABLE t(v VARIANT DEFAULT parse_json('1')) USING PARQUET")
+       sql("INSERT INTO t VALUES(DEFAULT)")
+-      checkAnswer(sql("select v from t"), sql("select parse_json('1')").collect())
++      // Native unshredding may use a different integer width for the same Variant value.
++      assert(sql("select v from t").collect().map(_.get(0).toString).toSeq == Seq("1"))
+     }
+   }
+
 diff --git a/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
 index 74cdee49e55..f7452c9abb7 100644
 --- a/sql/core/src/test/scala/org/apache/spark/sql/SQLQuerySuite.scala
@@ -1501,6 +1515,28 @@ index 8a0e2c29653..d276a51cbc6 100644
          }
        }
      }
+diff --git a/sql/core/src/test/scala/org/apache/spark/sql/VariantShreddingSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/VariantShreddingSuite.scala
+index fee375db10a..02a435c04e2 100644
+--- a/sql/core/src/test/scala/org/apache/spark/sql/VariantShreddingSuite.scala
++++ b/sql/core/src/test/scala/org/apache/spark/sql/VariantShreddingSuite.scala
+@@ -94,7 +94,16 @@ class VariantShreddingSuite extends QueryTest with SharedSparkSession with Parqu
+     spark.read.schema("v variant").parquet(path.getAbsolutePath)
+
+   def checkExpr(path: File, expr: String, expected: Any*): Unit = withAllParquetReaders {
+-    checkAnswer(read(path).selectExpr(expr), expected.map(Row(_)))
++    val df = read(path).selectExpr(expr)
++    if (df.schema.fields.head.dataType == VariantType) {
++      // Native unshredding may use different integer widths and metadata dictionaries.
++      // Compare values after collection; the other assertions check typed extraction.
++      val actual = df.collect().toSeq.map(row => Row(Option(row.get(0)).map(_.toString).orNull))
++      val rendered = expected.map(value => Row(Option(value).map(_.toString).orNull))
++      QueryTest.sameRows(rendered, actual).foreach(fail(_))
++    } else {
++      checkAnswer(df, expected.map(Row(_)))
++    }
+   }
+
+   def checkException(path: File, expr: String, msg: String): Unit = withAllParquetReaders {
 diff --git a/sql/core/src/test/scala/org/apache/spark/sql/collation/CollationSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/collation/CollationSuite.scala
 index 8f7a68bcbe6..88dbe1793c9 100644
 --- a/sql/core/src/test/scala/org/apache/spark/sql/collation/CollationSuite.scala

docs/source/user-guide/latest/datatypes.md

@@ -103,9 +103,18 @@ functions, and hashing a `CalendarInterval`. Remaining work is tracked by
 
 ## Variant
 
-| Type          | Status | Notes                                                                                                                                                                                                          |
-| ------------- | ------ | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
-| `VariantType` | 🔜     | Spark 4.0+. Native scan support is tracked by [#4295](https://github.com/apache/datafusion-comet/issues/4295); shredded Parquet read/write by [#3983](https://github.com/apache/datafusion-comet/issues/3983). |
+| Type          | Status | Notes                                                                                                                       |
+| ------------- | ------ | --------------------------------------------------------------------------------------------------------------------------- |
+| `VariantType` | ⚠️     | Spark 4.0+. Native Parquet scans support direct projection of top-level Variant columns, including missing-column defaults. |
+
+Direct projection requires `spark.sql.variant.allowReadingShredded=true` (the default in Spark
+4.1+), `spark.sql.variant.pushVariantIntoScan=false`, and the default Parquet timestamp inference
+settings. Nested Variant columns, pushed-down
+Variant field extraction, expressions, writes, shuffle and spill, Python operators, encrypted
+files, and Iceberg scans fall back to Spark. Spark also handles columnar-to-row conversion of
+the native scan output and strict reads with `allowReadingShredded=false`. Broader
+support is tracked by [#4295](https://github.com/apache/datafusion-comet/issues/4295) and
+[#3983](https://github.com/apache/datafusion-comet/issues/3983).
 
 ## Other
 

native/common/src/error.rs

@@ -21,6 +21,11 @@ use std::sync::Arc;
 
 #[derive(thiserror::Error, Debug, Clone)]
 pub enum SparkError {
+    #[error(
+        "[MALFORMED_VARIANT] Variant binary is malformed. Please check the data source is valid."
+    )]
+    MalformedVariant,
+
     // This list was generated from the Spark code. Many of the exceptions are not yet used by Comet
     #[error("[CAST_INVALID_INPUT] The value '{value}' of the type \"{from_type}\" cannot be cast to \"{to_type}\" \
         because it is malformed. Correct the value as per the syntax, or change its target type. \
@@ -301,6 +306,7 @@ impl SparkError {
     /// Get the error type name for JSON serialization
     pub(crate) fn error_type_name(&self) -> &'static str {
         match self {
+            SparkError::MalformedVariant => "MalformedVariant",
             SparkError::CastInvalidValue { .. } => "CastInvalidValue",
             SparkError::InvalidInputInCastToDatetime { .. } => "InvalidInputInCastToDatetime",
             SparkError::NumericValueOutOfRange { .. } => "NumericValueOutOfRange",
@@ -662,7 +668,8 @@ impl SparkError {
             | SparkError::InvalidIndexOfZero => "org/apache/spark/SparkArrayIndexOutOfBoundsException",
 
             // RuntimeException
-            SparkError::CannotParseDecimal
+            SparkError::MalformedVariant
+            | SparkError::CannotParseDecimal
             | SparkError::DuplicatedMapKey { .. }
             | SparkError::NullMapKey
             | SparkError::MapKeyValueDiffSizes
@@ -726,6 +733,7 @@ impl SparkError {
     /// Returns the Spark error class code for this error
     pub(crate) fn error_class(&self) -> Option<&'static str> {
         match self {
+            SparkError::MalformedVariant => Some("MALFORMED_VARIANT"),
             // Cast errors
             SparkError::CastInvalidValue { .. } => Some("CAST_INVALID_INPUT"),
             SparkError::InvalidInputInCastToDatetime { .. } => Some("CAST_INVALID_INPUT"),

native/core/src/execution/planner.rs

@@ -55,6 +55,7 @@ use arrow::datatypes::{
     DataType, Field, FieldRef, Fields, Schema, TimeUnit, DECIMAL128_MAX_PRECISION,
 };
 use arrow::ffi_stream::FFI_ArrowArrayStream;
+use arrow::record_batch::RecordBatch;
 use datafusion::functions_aggregate::bit_and_or_xor::{bit_and_udaf, bit_or_udaf, bit_xor_udaf};
 use datafusion::functions_aggregate::count::count_udaf;
 use datafusion::functions_aggregate::min_max::max_udaf;
@@ -109,8 +110,8 @@ use datafusion::datasource::listing::PartitionedFile;
 use datafusion::logical_expr::type_coercion::functions::fields_with_udf;
 use datafusion::logical_expr::type_coercion::other::get_coerce_type_for_case_expression;
 use datafusion::logical_expr::{
-    AggregateUDF, ReturnFieldArgs, ScalarUDF, TypeSignature, WindowFrame, WindowFrameBound,
-    WindowFrameUnits, WindowFunctionDefinition,
+    AggregateUDF, ColumnarValue, ReturnFieldArgs, ScalarUDF, TypeSignature, WindowFrame,
+    WindowFrameBound, WindowFrameUnits, WindowFunctionDefinition,
 };
 use datafusion::physical_expr::expressions::{Literal, StatsType};
 use datafusion::physical_expr::window::WindowExpr;
@@ -157,6 +158,7 @@ use jni::objects::{Global, JObject};
 use log::warn;
 use num::{BigInt, ToPrimitive};
 use object_store::path::Path;
+use parquet::variant::VariantType;
 use std::cmp::max;
 use std::{collections::HashMap, sync::Arc};
 
@@ -1000,6 +1002,39 @@ impl PhysicalPlanner {
         }
     }
 
+    /// Scan defaults are literals, except Variant's constant [value, metadata] storage struct.
+    /// Keep that exception here so general Variant expressions remain unsupported.
+    fn create_default_value(
+        &self,
+        spark_expr: &Expr,
+        input_schema: SchemaRef,
+        field: &Field,
+    ) -> Result<ScalarValue, ExecutionError> {
+        let expr = self.create_expr(spark_expr, Arc::clone(&input_schema))?;
+        if let Some(literal) = expr.downcast_ref::<DataFusionLiteral>() {
+            return Ok(literal.value().clone());
+        }
+        if !field.has_valid_extension_type::<VariantType>()
+            || expr.downcast_ref::<CreateNamedStruct>().is_none()
+            || expr.children().len() != 2
+            || !expr.children().iter().all(|child| {
+                child
+                    .downcast_ref::<DataFusionLiteral>()
+                    .is_some_and(|literal| matches!(literal.value(), ScalarValue::Binary(Some(_))))
+            })
+            || expr.data_type(&input_schema)? != *field.data_type()
+        {
+            return Err(GeneralError(
+                "Expected a literal or constant Variant storage struct for scan default"
+                    .to_string(),
+            ));
+        }
+        match expr.evaluate(&RecordBatch::new_empty(input_schema))? {
+            ColumnarValue::Scalar(value) => Ok(value),
+            _ => Err(GeneralError("Expected a scalar scan default".to_string())),
+        }
+    }
+
     /// Create a DataFusion physical sort expression from Spark physical expression
     fn create_sort_expr<'a>(
         &'a self,
@@ -1658,43 +1693,37 @@ impl PhysicalPlanner {
                             .collect()
                     };
 
-                let default_values: Option<HashMap<Column, ScalarValue>> = if !common
-                    .default_values
-                    .is_empty()
-                {
-                    // We have default values. Extract the two lists (same length) of values and
-                    // indexes in the schema, and then create a HashMap to use in the SchemaMapper.
-                    let default_values: Result<Vec<ScalarValue>, DataFusionError> = common
-                        .default_values
-                        .iter()
-                        .map(|expr| {
-                            let literal = self.create_expr(expr, Arc::clone(&required_schema))?;
-                            let df_literal =
-                                literal.downcast_ref::<DataFusionLiteral>().ok_or_else(|| {
-                                    GeneralError("Expected literal of default value.".to_string())
-                                })?;
-                            Ok(df_literal.value().clone())
-                        })
-                        .collect();
-                    let default_values = default_values?;
-                    let default_values_indexes: Vec<usize> = common
-                        .default_values_indexes
-                        .iter()
-                        .map(|offset| *offset as usize)
-                        .collect();
+                if common.default_values.len() != common.default_values_indexes.len() {
+                    return Err(GeneralError(
+                        "Scan default values and indexes have different lengths".to_string(),
+                    ));
+                }
+                let default_values = if common.default_values.is_empty() {
+                    None
+                } else {
                     Some(
-                        default_values_indexes
-                            .into_iter()
-                            .zip(default_values)
-                            .map(|(idx, scalar_value)| {
-                                let field = required_schema.field(idx);
-                                let column = Column::new(field.name().as_str(), idx);
-                                (column, scalar_value)
+                        common
+                            .default_values
+                            .iter()
+                            .zip(&common.default_values_indexes)
+                            .map(|(expr, offset)| {
+                                let idx = usize::try_from(*offset).map_err(|_| {
+                                    GeneralError(format!("Invalid scan default index {offset}"))
+                                })?;
+                                let field = required_schema.fields().get(idx).ok_or_else(|| {
+                                    GeneralError(format!(
+                                        "Scan default index {idx} is outside schema"
+                                    ))
+                                })?;
+                                let value = self.create_default_value(
+                                    expr,
+                                    Arc::clone(&required_schema),
+                                    field,
+                                )?;
+                                Ok((Column::new(field.name(), idx), value))
                             })
-                            .collect(),
+                            .collect::<Result<HashMap<_, _>, ExecutionError>>()?,
                     )
-                } else {
-                    None
                 };
 
                 // Get one file from this partition (we know it's not empty due to early return above)
@@ -5146,6 +5175,63 @@ mod tests {
         max_frame_size: usize,
     }
 
+    #[test]
+    fn variant_scan_default_requires_constant_storage() {
+        let planner = PhysicalPlanner::new(Arc::new(SessionContext::new()), 0);
+        let storage = DataType::Struct(Fields::from(vec![
+            Field::new("value", DataType::Binary, false),
+            Field::new("metadata", DataType::Binary, false),
+        ]));
+        let field = Field::new("v", storage.clone(), true).with_extension_type(VariantType);
+        let schema = Arc::new(Schema::new(vec![field.clone()]));
+        let bytes = |value| Expr {
+            expr_struct: Some(ExprStruct::Literal(spark_expression::Literal {
+                value: Some(literal::Value::BytesVal(value)),
+                datatype: Some(spark_expression::DataType {
+                    type_id: spark_expression::data_type::DataTypeId::Bytes as i32,
+                    type_info: None,
+                }),
+                is_null: false,
+            })),
+            ..Default::default()
+        };
+        let mut value = spark_expression::CreateNamedStruct {
+            names: vec!["value".to_string(), "metadata".to_string()],
+            values: vec![bytes(vec![0]), bytes(vec![1, 0, 0])],
+        };
+        let default_expr = |value| Expr {
+            expr_struct: Some(ExprStruct::CreateNamedStruct(value)),
+            ..Default::default()
+        };
+        let scalar = planner
+            .create_default_value(&default_expr(value.clone()), Arc::clone(&schema), &field)
+            .unwrap();
+        let ScalarValue::Struct(array) = scalar else {
+            panic!("expected a Variant storage scalar")
+        };
+        assert_eq!(array.data_type(), &storage);
+        assert_eq!(array.len(), 1);
+        assert_eq!(
+            ScalarValue::try_from_array(array.column(0).as_ref(), 0).unwrap(),
+            ScalarValue::Binary(Some(vec![0]))
+        );
+
+        // A struct expression is only a scan default for a marked Variant field.
+        let unmarked = Field::new("v", storage, true);
+        assert!(planner
+            .create_default_value(&default_expr(value.clone()), Arc::clone(&schema), &unmarked)
+            .is_err());
+        value.names.swap(0, 1);
+        assert!(planner
+            .create_default_value(&default_expr(value.clone()), Arc::clone(&schema), &field)
+            .is_err());
+        value.names.swap(0, 1);
+        value.values[0] = create_bound_reference(0);
+        assert!(planner
+            .create_default_value(&default_expr(value), schema, &field)
+            .is_err());
+    }
+
     #[test]
     fn spark_variant_schema_preserves_field_metadata() {
         let schema = convert_spark_types_to_arrow_schema(&[spark_operator::SparkStructField {