feat: Support native scans with unprojected Spark 4 VARIANT columns

spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala

@@ -540,15 +540,43 @@ case class CometScanRule(session: SparkSession)
             }
           }
 
-        // Comet serializes the whole table/scan schema to native, not just projected columns, so a
-        // type the native reader does not support (e.g. variant) breaks the scan even when that
-        // column is not projected. The readSchema allow-list only covers projected columns, so run
-        // the same allow-list over the full schema Comet may serialize. Reflection failure also
-        // falls back.
+        // The whole Iceberg table schema is serialized to native, but iceberg-rust can represent
+        // Variant in that schema as long as no projected field contains one. Match projected
+        // roots by field ID so historical snapshots still identify renamed columns, check them
+        // strictly, and allow Variant only under entirely unprojected roots. Other unsupported
+        // types still fail closed everywhere. An empty data projection is also strict because
+        // iceberg-rust currently interprets an empty field-id list as a request for every column.
         val schemaTypesSupported =
           try {
             val fullSchema = IcebergReflection.toSparkSchema(metadata.tableSchema)
-            typeChecker.isSchemaSupported(fullSchema, fallbackReasons)
+            val projectedDataColumns = scanExec.output.filterNot(_.isMetadataCol)
+            // DataTypeSupport recursively dispatches back to this override for struct fields,
+            // array elements, and map entries, so Variant is allowed at any nesting depth only
+            // when its entire top-level Iceberg field is unprojected.
+            val unprojectedTypeChecker = new CometScanTypeChecker() {
+              override def isTypeSupported(
+                  dt: DataType,
+                  name: String,
+                  reasons: ListBuffer[String]): Boolean =
+                isVariantType(dt) || super.isTypeSupported(dt, name, reasons)
+            }
+            val resolver = session.sessionState.conf.resolver
+            val tableFieldIds = IcebergReflection.buildFieldIdMapping(metadata.tableSchema)
+            val projectedFieldIds = projectedDataColumns.map { attr =>
+              metadata.globalFieldIdMapping.collectFirst {
+                case (fieldName, fieldId) if resolver(fieldName, attr.name) => fieldId
+              }
+            }
+            val resolvedProjectedFieldIds = projectedFieldIds.flatten.toSet
+            val hasUnresolvedProjectedFieldIds = projectedFieldIds.exists(_.isEmpty)
+
+            fullSchema.fields.forall { field =>
+              val isProjected = projectedDataColumns.isEmpty ||
+                hasUnresolvedProjectedFieldIds ||
+                tableFieldIds.get(field.name).forall(resolvedProjectedFieldIds.contains)
+              val checker = if (isProjected) typeChecker else unprojectedTypeChecker
+              checker.isTypeSupported(field.dataType, field.name, fallbackReasons)
+            }
           } catch {
             case e: Exception =>
               fallbackReasons += "Iceberg reflection failure: could not verify column " +
@@ -715,7 +743,7 @@ case class CometScanRule(session: SparkSession)
             true
         }
 
-        // Check for unsupported struct types in delete files
+        // Check for unsupported struct and Variant types in delete files
         val deleteFileTypesSupported = {
           var hasUnsupportedDeletes = false
 
@@ -766,12 +794,13 @@ case class CometScanRule(session: SparkSession)
                     }
                     fieldInfo match {
                       case Some((fieldName, fieldType)) =>
-                        if (fieldType.contains("struct")) {
+                        if (fieldType.contains("struct") || fieldType.equalsIgnoreCase(
+                            "variant")) {
                           hasUnsupportedDeletes = true
                           fallbackReasons +=
                             s"Equality delete on unsupported column type '$fieldName' " +
                               s"($fieldType) is not yet supported by iceberg-rust. " +
-                              "Struct types in equality deletes " +
+                              "Struct and Variant types in equality deletes " +
                               "require datum conversion support that is not yet implemented."
                         }
                       case None =>

spark/src/main/scala/org/apache/comet/serde/operator/CometNativeScan.scala

@@ -29,7 +29,7 @@ import org.apache.spark.sql.comet.{CometNativeExec, CometNativeScanExec, CometSc
 import org.apache.spark.sql.execution.{FileSourceScanExec, InSubqueryExec, SubqueryAdaptiveBroadcastExec}
 import org.apache.spark.sql.execution.datasources.parquet.ParquetUtils
 import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types.StructField
+import org.apache.spark.sql.types.{ArrayType, DataType, MapType, StructField, StructType}
 
 import org.apache.comet.{CometConf, ConfigEntry}
 import org.apache.comet.CometConf.COMET_EXEC_ENABLED
@@ -40,16 +40,26 @@ import org.apache.comet.serde.{CometOperatorSerde, Compatible, OperatorOuterClas
 import org.apache.comet.serde.ExprOuterClass.Expr
 import org.apache.comet.serde.OperatorOuterClass.Operator
 import org.apache.comet.serde.QueryPlanSerde.{exprToProto, serializeDataType}
+import org.apache.comet.shims.CometTypeShim
 
 /**
  * Validation and serde logic for Comet's native Parquet scan.
  */
-object CometNativeScan extends CometOperatorSerde[CometScanExec] with Logging {
+object CometNativeScan extends CometOperatorSerde[CometScanExec] with CometTypeShim with Logging {
 
   // DataFusion's table_partition_cols literal substitution matches by name, so a bare name
   // like "file_size" could collide with a real column of the same name. Prefix to avoid it.
   private val constantMetadataFieldPrefix = "_comet_metadata_"
 
+  private def containsVariantType(dataType: DataType): Boolean = dataType match {
+    case dt if isVariantType(dt) => true
+    case StructType(fields) => fields.exists(field => containsVariantType(field.dataType))
+    case ArrayType(elementType, _) => containsVariantType(elementType)
+    case MapType(keyType, valueType, _) =>
+      containsVariantType(keyType) || containsVariantType(valueType)
+    case _ => false
+  }
+
   /** Determine whether the scan is supported and tag the Spark plan with any fallback reasons */
   def isSupported(scanExec: FileSourceScanExec): Boolean = {
 
@@ -183,13 +193,28 @@ object CometNativeScan extends CometOperatorSerde[CometScanExec] with Logging {
         constantMetadataFields
       val partitionSchema = schema2Proto(partitionSchemaFields)
       val requiredSchema = schema2Proto(scan.requiredSchema)
-      val dataSchema = schema2Proto(scan.relation.dataSchema)
+
+      // Spark's required schema can prune a Variant column, including a Variant nested under an
+      // unrequested struct. The complete relation schema still contains that unsupported type,
+      // and serializing it would throw even though the native reader never needs those bytes.
+      // Keep ordinary fields unchanged and replace a requested Variant-bearing root with its
+      // already-validated, pruned required field. A requested actual Variant never reaches this
+      // point because CometScanRule keeps those scans on Spark.
+      val nativeDataSchema = StructType(scan.relation.dataSchema.fields.flatMap { field =>
+        if (containsVariantType(field.dataType)) {
+          scan.requiredSchema.fields.find(requiredField =>
+            scan.conf.resolver(requiredField.name, field.name))
+        } else {
+          Some(field)
+        }
+      })
+      val dataSchema = schema2Proto(nativeDataSchema)
 
       val dataSchemaIndexes = scan.requiredSchema.map(field => {
-        scan.relation.dataSchema.fieldIndex(field.name)
+        nativeDataSchema.fieldIndex(field.name)
       })
-      val partitionSchemaIndexes = scan.relation.dataSchema.fields.length until
-        (scan.relation.dataSchema.length + partitionSchemaFields.length)
+      val partitionSchemaIndexes = nativeDataSchema.fields.length until
+        (nativeDataSchema.length + partitionSchemaFields.length)
 
       val projectionVector = (dataSchemaIndexes ++ partitionSchemaIndexes).map(idx =>
         idx.toLong.asInstanceOf[java.lang.Long])

spark/src/test/resources/sql-tests/expressions/misc/variant.sql

@@ -22,14 +22,26 @@
 -- MinSparkVersion: 4.0
 
 statement
-CREATE TABLE test_variant(id INT, v VARIANT) USING parquet
+CREATE TABLE test_variant(id INT, v VARIANT, tail STRING) USING parquet
 
 statement
 INSERT INTO test_variant VALUES
-  (1, parse_json('{"a": 1, "b": "hello"}')),
-  (2, parse_json('{"a": 2, "b": "world"}')),
-  (3, parse_json('null')),
-  (4, NULL)
+  (1, parse_json('{"a": 1, "b": "hello"}'), 'first'),
+  (2, parse_json('{"a": 2, "b": "world"}'), NULL),
+  (3, parse_json('null'), 'variant-null'),
+  (4, NULL, 'sql-null')
+
+-- A plain Parquet scan can remain native when its required schema prunes the
+-- Variant column completely, including both SQL NULL and Variant null values.
+query
+SELECT id FROM test_variant ORDER BY id
+
+-- A projected column after the pruned Variant must use its rebased native index.
+query
+SELECT tail FROM test_variant ORDER BY id
+
+query
+SELECT id, tail FROM test_variant WHERE tail IS NOT NULL ORDER BY id
 
 query expect_fallback(type VariantType)
 SELECT id, v FROM test_variant ORDER BY id
@@ -44,12 +56,85 @@ query expect_fallback(type VariantType)
 SELECT COUNT(*) FROM test_variant WHERE v IS NOT NULL
 
 statement
-CREATE TABLE test_variant_struct(id INT, s STRUCT<v: VARIANT>) USING parquet
+CREATE TABLE test_variant_struct(id INT, s STRUCT<safe: INT, v: VARIANT>, tail STRING)
+USING parquet
 
 statement
 INSERT INTO test_variant_struct VALUES
-  (1, named_struct('v', parse_json('{"x": 10}'))),
-  (2, named_struct('v', parse_json('{"x": 20}')))
+  (1, named_struct('safe', 10, 'v', parse_json('{"x": 10}')), 'first'),
+  (2, named_struct('safe', NULL, 'v', parse_json('{"x": 20}')), NULL),
+  (3, NULL, 'null-parent')
+
+query
+SELECT id FROM test_variant_struct ORDER BY id
+
+query
+SELECT tail FROM test_variant_struct ORDER BY id
+
+-- Projecting a supported sibling replaces the full struct with Spark's pruned nested schema.
+query
+SELECT s.safe FROM test_variant_struct ORDER BY id
 
 query expect_fallback(type VariantType)
 SELECT id, s FROM test_variant_struct ORDER BY id
+
+statement
+CREATE TABLE test_variant_collections(
+  id INT,
+  variants ARRAY<VARIANT>,
+  variants_by_key MAP<STRING, VARIANT>,
+  tail STRING)
+USING parquet
+
+statement
+INSERT INTO test_variant_collections VALUES
+  (1,
+   array(parse_json('{"x": 1}'), parse_json('null')),
+   map('first', parse_json('{"x": 2}')),
+   'first'),
+  (2,
+   array(CAST(NULL AS VARIANT)),
+   map('sql-null', CAST(NULL AS VARIANT)),
+   NULL),
+  (3, NULL, NULL, 'null-collections')
+
+-- Variant-bearing arrays and maps can be pruned as entire top-level fields.
+query
+SELECT id, tail FROM test_variant_collections ORDER BY id
+
+-- Exposing either collection still requires Spark to decode its nested Variant values.
+query expect_fallback(type VariantType)
+SELECT id, variants FROM test_variant_collections ORDER BY id
+
+query expect_fallback(type VariantType)
+SELECT id, variants_by_key FROM test_variant_collections ORDER BY id
+
+statement
+CREATE TABLE test_variant_partitioned(id INT, v VARIANT, tail STRING, p INT)
+USING parquet PARTITIONED BY (p)
+
+statement
+INSERT INTO test_variant_partitioned VALUES
+  (1, parse_json('{"a": 1}'), 'first', 10),
+  (2, NULL, 'second', 20)
+
+-- Partition columns are appended after the pruned data schema, so their offsets must be rebased.
+query
+SELECT tail, p FROM test_variant_partitioned ORDER BY id
+
+-- File-constant metadata follows the partition columns and needs the same rebased offsets.
+query
+SELECT tail, p, _metadata.file_name FROM test_variant_partitioned ORDER BY id
+
+statement
+CREATE TABLE test_plain_variant_shape(id INT, payload STRUCT<value: BINARY, metadata: BINARY>)
+USING parquet
+
+statement
+INSERT INTO test_plain_variant_shape VALUES
+  (1, named_struct('value', X'01', 'metadata', X'02')),
+  (2, NULL)
+
+-- An ordinary binary struct with Variant-like field names is not a logical Variant.
+query
+SELECT payload FROM test_plain_variant_shape ORDER BY id