feat: add native Variant null and validity predicates

docs/source/contributor-guide/expression-audits/misc_funcs.md

@@ -37,6 +37,24 @@
 
 - Resolved to a literal by the analyzer; same as `user`.
 
+## is_valid_variant
+
+- Spark 3.4.3 (audited 2026-10-04): unavailable.
+- Spark 3.5.8 (audited 2026-10-04): unavailable.
+- Spark 4.0.1 (audited 2026-10-04): unavailable.
+- Spark 4.1.1 (audited 2026-10-04): unavailable.
+- Spark 4.2.0 (audited 2026-10-04): SQL NULL propagates. Validates the metadata version and every reachable value and object key. It accepts unused malformed metadata, invalid UTF-8, trailing bytes, and unordered object fields. Native validation follows Spark's accessors rather than Arrow's stricter format validation.
+- Comet accepts top-level Variant columns and literals in projections and filters. Unshredded scans preserve raw payloads for predicate validation, matching Spark's constructor checks for missing children, metadata version and size. `CometVariantProjectionSuite` covers malformed scan bytes and constructor errors (verified 2026-10-05).
+
+## is_variant_null
+
+- Spark 3.4.3 (audited 2026-10-04): unavailable.
+- Spark 3.5.8 (audited 2026-10-04): unavailable.
+- Spark 4.0.1 (audited 2026-10-04): returns false for SQL NULL. An empty value raises `MALFORMED_VARIANT`; otherwise only a zero first value byte returns true. The remaining value bytes and metadata are not validated.
+- Spark 4.1.1 (audited 2026-10-04): identical to 4.0.1.
+- Spark 4.2.0 (audited 2026-10-04): identical to 4.0.1.
+- Comet accepts top-level Variant columns and literals in projections and filters and retains generic Variant expression fallback.
+
 ## monotonically_increasing_id
 
 - Spark 3.4.3 (audited 2026-05-27): byte-for-byte identical to 4.1.1. `MonotonicallyIncreasingID() extends LeafExpression with Stateful`; produces a Long that encodes the partition id in the upper 31 bits and a per-partition row counter in the lower 33 bits. Comet emits an empty `MonotonicallyIncreasingId` proto and the native side produces the same encoding.

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

@@ -112,13 +112,19 @@ Direct projection requires explicit configuration on every supported Spark versi
 `spark.sql.variant.pushVariantIntoScan=false` (defaults to true in Spark 4.1+), with the default
 Parquet timestamp inference settings. Support for Spark's whole-value pushdown rewrite is tracked
 by [#5519](https://github.com/apache/datafusion-comet/issues/5519). Nested Variant columns, pushed-down
-Variant field extraction, expressions, writes, shuffle and spill, Python operators, encrypted
+Variant field extraction, writes, shuffle and spill, Python operators, encrypted
 files, and Iceberg scans that read a Variant column fall back to Spark. Iceberg scans of tables
 whose Variant columns the query does not read run natively. 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).
 
+Projections and filters support `is_variant_null` (Spark 4.0+) and `is_valid_variant` (Spark 4.2+)
+on top-level Variant columns and literals. Other Variant expressions fall back. The predicates
+preserve Spark's distinct handling of SQL NULL, Variant null, and malformed bytes. Unshredded
+scans pass payloads to the predicates after checking required children, metadata version and
+Spark's size limit. Shredded inputs still require reconstruction before evaluation.
+
 Shredded reconstruction can be slower than Spark's reader; see the
 [focused scan and allocation measurements in PR #5868](https://github.com/apache/datafusion-comet/pull/5868).
 

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

@@ -508,7 +508,8 @@ to Spark ([#2837](https://github.com/apache/datafusion-comet/issues/2837)). Enab
 | `equal_null` | ✅ | — | Lowers to `<=>` (`EqualNullSafe`) |
 | `hll_sketch_estimate` | ✅ | Native | Spark 4.0+ only; falls back by default, the native path is opt-in via allowIncompatible ([details](compatibility/expressions/misc.md)) |
 | `hll_union` | ✅ | Native | Spark 4.0+ only; falls back by default, the native path is opt-in via allowIncompatible ([details](compatibility/expressions/misc.md)) |
-| `is_variant_null` | 🔜 | — | Requires `VariantType` support |
+| `is_valid_variant` | ✅ | Native | Spark 4.2+. Top-level Variant columns and literals. |
+| `is_variant_null` | ✅ | Native | Spark 4.0+. Top-level Variant columns and literals. |
 | `monotonically_increasing_id` | ✅ | Native |  |
 | `parse_json` | 🔜 | — | Requires `VariantType` support |
 | `raise_error` | 🔜 | — | Raises a runtime error |

native/common/src/error.rs

@@ -26,6 +26,11 @@ pub enum SparkError {
     )]
     MalformedVariant,
 
+    #[error(
+        "[VARIANT_CONSTRUCTOR_SIZE_LIMIT] Variant value or metadata exceeds Spark's size limit."
+    )]
+    VariantConstructorSizeLimit,
+
     // 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. \
@@ -332,6 +337,7 @@ impl SparkError {
     pub(crate) fn error_type_name(&self) -> &'static str {
         match self {
             SparkError::MalformedVariant => "MalformedVariant",
+            SparkError::VariantConstructorSizeLimit => "VariantConstructorSizeLimit",
             SparkError::CastInvalidValue { .. } => "CastInvalidValue",
             SparkError::InvalidInputInCastToDatetime { .. } => "InvalidInputInCastToDatetime",
             SparkError::NumericValueOutOfRange { .. } => "NumericValueOutOfRange",
@@ -707,6 +713,7 @@ impl SparkError {
 
             // RuntimeException
             SparkError::MalformedVariant
+            | SparkError::VariantConstructorSizeLimit
             | SparkError::CannotParseDecimal
             | SparkError::DuplicatedMapKey { .. }
             | SparkError::NullMapKey
@@ -778,6 +785,7 @@ impl SparkError {
     pub(crate) fn error_class(&self) -> Option<&'static str> {
         match self {
             SparkError::MalformedVariant => Some("MALFORMED_VARIANT"),
+            SparkError::VariantConstructorSizeLimit => Some("VARIANT_CONSTRUCTOR_SIZE_LIMIT"),
             // Cast errors
             SparkError::CastInvalidValue { .. } => Some("CAST_INVALID_INPUT"),
             SparkError::InvalidInputInCastToDatetime { .. } => Some("CAST_INVALID_INPUT"),

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

@@ -670,6 +670,7 @@ fn parquet_probe(
         false,
         false,
         false,
+        128 * 1024 * 1024,
     )
     .unwrap();
     (file, scan)
@@ -775,6 +776,7 @@ async fn reader_filter_crosses_null_check_conjunction_and_retains_residual() {
             false,
             false,
             false,
+            128 * 1024 * 1024,
         )
         .unwrap();
         let checks = [("key", 0), ("payload", 1), ("other", 2)].map(|(name, index)| {

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

@@ -86,6 +86,7 @@ fn scan(
         false,
         false,
         false,
+        128 * 1024 * 1024,
     )
     .unwrap()
 }

native/core/src/execution/operators/dynamic_filter/join/tests/schema_errors/partition_columns.rs

@@ -71,6 +71,7 @@ fn partitioned_scan(
         false,
         false,
         false,
+        128 * 1024 * 1024,
     )
     .unwrap()
 }

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

@@ -105,6 +105,7 @@ async fn assert_timestamp_overflow_preserved(nested: bool) {
             false,
             false,
             false,
+            128 * 1024 * 1024,
         )
         .unwrap();
         let join = single_key_join_plans(

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

@@ -196,6 +196,7 @@ fn parquet_scan(
         false,
         false,
         false,
+        128 * 1024 * 1024,
     )
     .unwrap()
 }

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

@@ -95,6 +95,7 @@ fn timestamp_input(
         false,
         false,
         false,
+        128 * 1024 * 1024,
     )
     .unwrap();
     (file, scan)

native/core/src/execution/planner.rs

@@ -570,7 +570,9 @@ impl PhysicalPlanner {
                 )))
             }
             ExprStruct::Literal(literal) => {
-                let data_type = to_arrow_datatype(literal.datatype.as_ref().unwrap());
+                let literal_field =
+                    to_arrow_field("lit", literal.datatype.as_ref().unwrap(), literal.is_null);
+                let data_type = literal_field.data_type().clone();
                 let scalar_value = if literal.is_null {
                     match data_type {
                         DataType::Boolean => ScalarValue::Boolean(None),
@@ -648,6 +650,16 @@ impl PhysicalPlanner {
                         Value::DoubleVal(value) => ScalarValue::Float64(Some(*value)),
                         Value::StringVal(value) => ScalarValue::Utf8(Some(value.clone())),
                         Value::BytesVal(value) => ScalarValue::Binary(Some(value.clone())),
+                        Value::VariantVal(value) => {
+                            if literal.datatype.as_ref().unwrap().type_id != spark_expression::data_type::DataTypeId::Variant as i32 {
+                                return Err(GeneralError("Variant literal requires Variant datatype".to_string()));
+                            }
+                            let DataType::Struct(fields) = data_type else { unreachable!() };
+                            ScalarStructBuilder::new()
+                                .with_scalar(Arc::clone(&fields[0]), ScalarValue::Binary(Some(value.value.clone())))
+                                .with_scalar(Arc::clone(&fields[1]), ScalarValue::Binary(Some(value.metadata.clone())))
+                                .build()?
+                        }
                         Value::DecimalVal(value) => {
                             let big_integer = BigInt::from_signed_bytes_be(value);
                             let integer = big_integer.to_i128().ok_or_else(|| {
@@ -678,7 +690,10 @@ impl PhysicalPlanner {
                         }
                     }
                 };
-                Ok(Arc::new(DataFusionLiteral::new(scalar_value)))
+                Ok(Arc::new(DataFusionLiteral::new_with_metadata(
+                    scalar_value,
+                    Some(literal_field.metadata().into()),
+                )))
             }
             ExprStruct::Cast(expr) => {
                 let child = self.create_expr(expr.child.as_ref().unwrap(), input_schema)?;
@@ -1864,6 +1879,7 @@ impl PhysicalPlanner {
                     common.encryption_enabled,
                     common.use_field_id,
                     common.require_field_ids,
+                    common.variant_size_limit as usize,
                 )?;
                 Ok((
                     vec![],
@@ -3699,6 +3715,14 @@ impl PhysicalPlanner {
             .collect::<Result<Vec<_>, _>>()?;
 
         let fun_name = &expr.func;
+        if matches!(fun_name.as_str(), "is_variant_null" | "is_valid_variant")
+            && (args.len() != 1
+                || !args[0]
+                    .return_field(&input_schema)?
+                    .has_valid_extension_type::<parquet::variant::VariantType>())
+        {
+            return Err(GeneralError(format!("{fun_name} requires a Variant input")));
+        }
         // `map_entries` needs its argument's entry `value` field widened to nullable first (only
         // that outer field). See `widen_map_entry_value_nullable`.
         let args = if fun_name == "map_entries" {
@@ -5098,6 +5122,77 @@ mod tests {
         assert_eq!(field.metadata().get("source"), Some(&"spark".to_string()));
     }
 
+    #[test]
+    fn variant_predicates_require_logical_variant_inputs() {
+        let planner = PhysicalPlanner::new(Arc::new(SessionContext::new()), 0);
+        let datatype = spark_expression::DataType {
+            type_id: spark_expression::data_type::DataTypeId::Variant as i32,
+            type_info: None,
+        };
+        let field = super::to_arrow_field("v", &datatype, true);
+        let bound = Expr {
+            expr_struct: Some(Bound(spark_expression::BoundReference {
+                index: 0,
+                datatype: Some(datatype.clone()),
+            })),
+            ..Default::default()
+        };
+        for name in ["is_variant_null", "is_valid_variant"] {
+            let predicate = |arg| spark_expression::ScalarFunc {
+                func: name.to_string(),
+                args: vec![arg],
+                return_type: Some(spark_expression::DataType {
+                    type_id: spark_expression::data_type::DataTypeId::Bool as i32,
+                    type_info: None,
+                }),
+                ..Default::default()
+            };
+            for (field, expected) in [
+                (field.clone(), true),
+                (Field::new("v", field.data_type().clone(), true), false),
+                (
+                    field
+                        .clone()
+                        .with_metadata(std::collections::HashMap::from([(
+                            "ARROW:extension:name".to_string(),
+                            "example.variant".to_string(),
+                        )])),
+                    false,
+                ),
+            ] {
+                let result = planner.create_scalar_function_expr(
+                    &predicate(bound.clone()),
+                    Arc::new(Schema::new(vec![field])),
+                );
+                assert_eq!(result.is_ok(), expected, "{name}: {result:?}");
+            }
+            for is_null in [false, true] {
+                let input = Expr {
+                    expr_struct: Some(ExprStruct::Literal(spark_expression::Literal {
+                        datatype: Some(datatype.clone()),
+                        is_null,
+                        value: Some(literal::Value::VariantVal(
+                            spark_expression::VariantLiteral {
+                                value: vec![0],
+                                metadata: vec![1, 0, 0],
+                            },
+                        )),
+                    })),
+                    ..Default::default()
+                };
+                let schema = Arc::new(Schema::empty());
+                let literal = planner.create_expr(&input, Arc::clone(&schema)).unwrap();
+                assert!(literal
+                    .return_field(&schema)
+                    .unwrap()
+                    .has_valid_extension_type::<VariantType>());
+                assert!(planner
+                    .create_scalar_function_expr(&predicate(input), schema)
+                    .is_ok());
+            }
+        }
+    }
+
     fn create_sort_order(index: i32, type_id: i32, descending: bool, nulls_first: bool) -> Expr {
         Expr {
             expr_struct: Some(SortOrder(Box::new(spark_expression::SortOrder {

native/core/src/parquet/cast_column.rs

@@ -288,6 +288,14 @@ impl PhysicalExpr for CometCastColumnExpr {
                 ColumnarValue::Array(array) => Ok(ColumnarValue::Array(normalize_variant_array(
                     &array,
                     &self.target_field,
+                    self.parquet_options
+                        .as_ref()
+                        .ok_or_else(|| {
+                            DataFusionError::Plan(
+                                "Variant scan requires Parquet options".to_string(),
+                            )
+                        })?
+                        .variant_size_limit,
                 )?)),
                 ColumnarValue::Scalar(_) => Err(DataFusionError::Execution(
                     "Variant Parquet projection requires an array".to_string(),

native/core/src/parquet/cast_column/variant.rs

@@ -41,6 +41,7 @@ use std::{
 pub(super) fn normalize_variant_array(
     array: &ArrayRef,
     target_field: &FieldRef,
+    size_limit: usize,
 ) -> DataFusionResult<ArrayRef> {
     let DataType::Struct(fields) = target_field.data_type() else {
         return Err(DataFusionError::Execution(
@@ -64,6 +65,41 @@ pub(super) fn normalize_variant_array(
     // shredded output is then rebuilt with Spark's byte encoding.
     let array = normalize_variant_storage(array)?;
     let variant = VariantArray::try_new(array.as_ref())?;
+    if variant.typed_value_column().is_none() {
+        // Spark's ShreddingUtils.rebuild passes unshredded bytes through. Validate the
+        // required children and metadata version, leaving payload validation to consumers
+        // such as is_valid_variant (https://github.com/apache/datafusion-comet/issues/5429).
+        let mut value = cast(variant.value_column().as_ref(), &DataType::Binary)?;
+        let metadata = cast(variant.metadata_column().as_ref(), &DataType::Binary)?;
+        for row in 0..variant.len() {
+            if variant.inner().is_valid(row)
+                && (value.is_null(row)
+                    || metadata.is_null(row)
+                    || metadata
+                        .as_binary::<i32>()
+                        .value(row)
+                        .first()
+                        .is_none_or(|b| b & 15 != 1))
+            {
+                return Err(SparkError::MalformedVariant.into());
+            }
+            if variant.inner().is_valid(row)
+                && (value.as_binary::<i32>().value(row).len() > size_limit
+                    || metadata.as_binary::<i32>().value(row).len() > size_limit)
+            {
+                return Err(SparkError::VariantConstructorSizeLimit.into());
+            }
+        }
+        if variant.inner().null_count() != 0 {
+            value =
+                arrow::compute::nullif(value.as_ref(), &arrow::compute::is_null(variant.inner())?)?;
+        }
+        return Ok(Arc::new(StructArray::try_new(
+            fields.clone(),
+            vec![value, metadata],
+            variant.inner().nulls().cloned(),
+        )?));
+    }
     let normalize = |metadata: Option<&ArrayRef>| -> DataFusionResult<ArrayRef> {
         let extended = extend_shredded_metadata(&variant, metadata)?;
         let prepared = prepare_variant_for_unshredding(&variant, extended.as_ref().or(metadata))?;
@@ -73,25 +109,11 @@ pub(super) fn normalize_variant_array(
             }
             error => error.into(),
         })?;
-        let (value, metadata) = if variant.typed_value_column().is_some() {
-            rebuild_spark_variant(
-                &variant,
-                unshredded.value_column(),
-                unshredded.metadata_column(),
-            )?
-        } else {
-            // Spark passes unshredded bytes through, including dictionary order, unused keys,
-            // and wide scalar encodings. Preparation above still validates legacy input.
-            let mut value = cast(variant.value_column().as_ref(), &DataType::Binary)?;
-            let metadata = cast(variant.metadata_column().as_ref(), &DataType::Binary)?;
-            if variant.inner().null_count() != 0 {
-                value = arrow::compute::nullif(
-                    value.as_ref(),
-                    &arrow::compute::is_null(variant.inner())?,
-                )?;
-            }
-            (value, metadata)
-        };
+        let (value, metadata) = rebuild_spark_variant(
+            &variant,
+            unshredded.value_column(),
+            unshredded.metadata_column(),
+        )?;
         Ok(Arc::new(StructArray::try_new(
             fields.clone(),
             vec![value, metadata],