From ebd3711c6953e3239f08140420e56aa338743150 Mon Sep 17 00:00:00 2001 From: Chao Sun Date: Fri, 14 Aug 2026 22:35:28 -0700 Subject: [PATCH 1/5] fix: keep scans native when Variant columns are unprojected --- .../apache/comet/rules/CometScanRule.scala | 34 +++-- .../serde/operator/CometNativeScan.scala | 37 ++++- .../sql-tests/expressions/misc/variant.sql | 70 ++++++++-- .../comet/CometIcebergNativeSuite.scala | 127 +++++++++++++++++- 4 files changed, 244 insertions(+), 24 deletions(-) diff --git a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala index 2c15bb5e4e7..0b8da351fe5 100644 --- a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala +++ b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala @@ -540,15 +540,30 @@ 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. Check projected + // roots 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) + 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 + + fullSchema.fields.forall { field => + val isProjected = projectedDataColumns.isEmpty || + projectedDataColumns.exists(attr => resolver(attr.name, field.name)) + 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 +730,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 +781,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 => diff --git a/spark/src/main/scala/org/apache/comet/serde/operator/CometNativeScan.scala b/spark/src/main/scala/org/apache/comet/serde/operator/CometNativeScan.scala index 0240bd4a076..e395ac6d9d3 100644 --- a/spark/src/main/scala/org/apache/comet/serde/operator/CometNativeScan.scala +++ b/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]) diff --git a/spark/src/test/resources/sql-tests/expressions/misc/variant.sql b/spark/src/test/resources/sql-tests/expressions/misc/variant.sql index 72ea174edf8..6b30a2358db 100644 --- a/spark/src/test/resources/sql-tests/expressions/misc/variant.sql +++ b/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,54 @@ 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) USING parquet +CREATE TABLE test_variant_struct(id INT, s STRUCT, 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_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) +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 diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala index 362f843ccaf..419d546d337 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala @@ -28,7 +28,7 @@ import scala.jdk.CollectionConverters._ import org.apache.spark.CometListenerBusUtils import org.apache.spark.scheduler.{SparkListener, SparkListenerTaskEnd} -import org.apache.spark.sql.{CometTestBase, DataFrame} +import org.apache.spark.sql.{CometTestBase, DataFrame, Row} import org.apache.spark.sql.catalyst.expressions.DynamicPruningExpression import org.apache.spark.sql.comet._ import org.apache.spark.sql.comet.execution.shuffle.CometShuffleExchangeExec @@ -5160,6 +5160,131 @@ class CometIcebergNativeSuite } } + test("unprojected variant columns do not disable native Iceberg scans") { + assume(isSpark40Plus, "VARIANT type requires Spark 4.0+") + assume(icebergAvailable, "Iceberg not available in classpath") + assume(icebergVersionAtLeast(1, 10), "VARIANT type requires Iceberg 1.10+") + withTempIcebergDir { warehouseDir => + withSQLConf( + "spark.sql.catalog.test_cat" -> "org.apache.iceberg.spark.SparkCatalog", + "spark.sql.catalog.test_cat.type" -> "hadoop", + "spark.sql.catalog.test_cat.warehouse" -> warehouseDir.getAbsolutePath, + CometConf.COMET_ENABLED.key -> "true", + CometConf.COMET_EXEC_ENABLED.key -> "true", + CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") { + val table = "test_cat.db.variant_projection" + try { + spark.sql( + s"CREATE TABLE $table (id BIGINT, label STRING, data VARIANT) USING iceberg " + + "TBLPROPERTIES ('format-version' = '3')") + spark.sql(s""" + INSERT INTO $table VALUES + (1, 'object', parse_json('{"num": 25}')), + (2, NULL, parse_json('null')), + (NULL, 'sql-null', NULL), + (4, 'array', parse_json('[1, 2]')) + """) + + // Iceberg 1.10's Spark reader dereferences a null requested type when it encounters an + // unprojected, annotated Variant. Keep Variant projected for the Spark reference read, + // then discard it from the expected rows before checking the native pruned projection. + val sparkAllRows = withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + spark + .sql(s"SELECT id, label, data FROM $table ORDER BY id NULLS FIRST") + .collect() + .map(row => Row(row.get(0), row.get(1))) + .toSeq + } + assert( + sparkAllRows == + Seq(Row(null, "sql-null"), Row(1L, "object"), Row(2L, null), Row(4L, "array"))) + val allRows = spark.sql(s"SELECT id, label FROM $table ORDER BY id NULLS FIRST") + checkCometAnswer(allRows, sparkAllRows) + assertSingleNativeScan(allRows.queryExecution.executedPlan) + + val sparkFilteredRows = withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + spark + .sql(s"SELECT id, data FROM $table " + + "WHERE label IS NOT NULL ORDER BY id NULLS FIRST") + .collect() + .map(row => Row(row.get(0))) + .toSeq + } + val filteredRows = + spark.sql(s"SELECT id FROM $table WHERE label IS NOT NULL ORDER BY id NULLS FIRST") + checkCometAnswer(filteredRows, sparkFilteredRows) + assertSingleNativeScan(filteredRows.queryExecution.executedPlan) + + checkIcebergNativeScanFallback( + s"SELECT id FROM $table WHERE " + + "try_variant_get(data, '$.num', 'int') > 20 ORDER BY id", + "projected VARIANT columns remain unsupported") + + withSQLConf("spark.sql.iceberg.aggregate-push-down.enabled" -> "false") { + val emptyProjection = spark.sql(s"SELECT COUNT(*) FROM $table") + assert( + collectIcebergNativeScans(emptyProjection.queryExecution.executedPlan).isEmpty, + "An empty projection must not read every field from a VARIANT-bearing table") + } + + val metadataOnly = spark.sql(s"SELECT _file FROM $table") + assert( + collectIcebergNativeScans(metadataOnly.queryExecution.executedPlan).isEmpty, + "A metadata-only projection must not read a VARIANT-bearing data schema") + } finally { + spark.sql(s"DROP TABLE IF EXISTS $table PURGE") + } + } + } + } + + test("projecting a struct containing an unprojected variant still falls back") { + assume(isSpark40Plus, "VARIANT type requires Spark 4.0+") + assume(icebergAvailable, "Iceberg not available in classpath") + assume(icebergVersionAtLeast(1, 10), "VARIANT type requires Iceberg 1.10+") + withTempIcebergDir { warehouseDir => + withSQLConf( + "spark.sql.catalog.test_cat" -> "org.apache.iceberg.spark.SparkCatalog", + "spark.sql.catalog.test_cat.type" -> "hadoop", + "spark.sql.catalog.test_cat.warehouse" -> warehouseDir.getAbsolutePath, + CometConf.COMET_ENABLED.key -> "true", + CometConf.COMET_EXEC_ENABLED.key -> "true", + CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") { + val table = "test_cat.db.nested_variant_projection" + try { + spark.sql( + s"CREATE TABLE $table " + + "(id BIGINT, nested STRUCT) USING iceberg " + + "TBLPROPERTIES ('format-version' = '3')") + spark.sql(s""" + INSERT INTO $table VALUES + (1, named_struct('label', 'first', 'data', parse_json('{"num": 1}'))), + (2, named_struct('label', NULL, 'data', NULL)), + (3, NULL) + """) + + val sparkScalarRows = withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + spark + .sql(s"SELECT id, nested FROM $table ORDER BY id") + .collect() + .map(row => Row(row.get(0))) + .toSeq + } + val scalarRows = spark.sql(s"SELECT id FROM $table ORDER BY id") + checkCometAnswer(scalarRows, sparkScalarRows) + assertSingleNativeScan(scalarRows.queryExecution.executedPlan) + + val nestedProjection = spark.sql(s"SELECT nested.label FROM $table ORDER BY id") + assert( + collectIcebergNativeScans(nestedProjection.queryExecution.executedPlan).isEmpty, + "iceberg-rust rejects a projected parent containing a VARIANT field") + } finally { + spark.sql(s"DROP TABLE IF EXISTS $table PURGE") + } + } + } + } + test("partition evolution - _partition contains fields from all historical specs") { assume(icebergAvailable, "Iceberg not available in classpath") From 902ed1107e46003804533791e8f437ebcd67416f Mon Sep 17 00:00:00 2001 From: Chao Sun Date: Sat, 15 Aug 2026 15:27:17 -0700 Subject: [PATCH 2/5] fix: preserve Variant fallback for renamed Iceberg columns --- .../apache/comet/rules/CometScanRule.scala | 16 ++++++++-- .../comet/CometIcebergNativeSuite.scala | 29 +++++++++++++++---- 2 files changed, 36 insertions(+), 9 deletions(-) diff --git a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala index 0b8da351fe5..ebe15a3a606 100644 --- a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala +++ b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala @@ -541,8 +541,9 @@ case class CometScanRule(session: SparkSession) } // 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. Check projected - // roots strictly and allow Variant only under entirely unprojected roots. Other unsupported + // 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 = @@ -557,10 +558,19 @@ case class CometScanRule(session: SparkSession) 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 || - projectedDataColumns.exists(attr => resolver(attr.name, field.name)) + hasUnresolvedProjectedFieldIds || + tableFieldIds.get(field.name).forall(resolvedProjectedFieldIds.contains) val checker = if (isProjected) typeChecker else unprojectedTypeChecker checker.isTypeSupported(field.dataType, field.name, fallbackReasons) } diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala index 419d546d337..c3bb25e10e4 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala @@ -5188,8 +5188,9 @@ class CometIcebergNativeSuite // Iceberg 1.10's Spark reader dereferences a null requested type when it encounters an // unprojected, annotated Variant. Keep Variant projected for the Spark reference read, // then discard it from the expected rows before checking the native pruned projection. - val sparkAllRows = withSQLConf(CometConf.COMET_ENABLED.key -> "false") { - spark + var sparkAllRows = Seq.empty[Row] + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + sparkAllRows = spark .sql(s"SELECT id, label, data FROM $table ORDER BY id NULLS FIRST") .collect() .map(row => Row(row.get(0), row.get(1))) @@ -5202,8 +5203,9 @@ class CometIcebergNativeSuite checkCometAnswer(allRows, sparkAllRows) assertSingleNativeScan(allRows.queryExecution.executedPlan) - val sparkFilteredRows = withSQLConf(CometConf.COMET_ENABLED.key -> "false") { - spark + var sparkFilteredRows = Seq.empty[Row] + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + sparkFilteredRows = spark .sql(s"SELECT id, data FROM $table " + "WHERE label IS NOT NULL ORDER BY id NULLS FIRST") .collect() @@ -5263,8 +5265,9 @@ class CometIcebergNativeSuite (3, NULL) """) - val sparkScalarRows = withSQLConf(CometConf.COMET_ENABLED.key -> "false") { - spark + var sparkScalarRows = Seq.empty[Row] + withSQLConf(CometConf.COMET_ENABLED.key -> "false") { + sparkScalarRows = spark .sql(s"SELECT id, nested FROM $table ORDER BY id") .collect() .map(row => Row(row.get(0))) @@ -5278,6 +5281,20 @@ class CometIcebergNativeSuite assert( collectIcebergNativeScans(nestedProjection.queryExecution.executedPlan).isEmpty, "iceberg-rust rejects a projected parent containing a VARIANT field") + + val snapshotId = spark + .sql(s"SELECT snapshot_id FROM $table.snapshots ORDER BY committed_at DESC LIMIT 1") + .collect() + .head + .getLong(0) + spark.sql(s"ALTER TABLE $table RENAME COLUMN nested TO renamed") + + val historicalNestedProjection = + spark.sql(s"SELECT nested.label FROM $table VERSION AS OF $snapshotId ORDER BY id") + assert( + collectIcebergNativeScans( + historicalNestedProjection.queryExecution.executedPlan).isEmpty, + "A renamed parent containing a VARIANT field must fall back for historical snapshots") } finally { spark.sql(s"DROP TABLE IF EXISTS $table PURGE") } From f690faeb2cd3f8b3c6adcce5f80e141960d67f53 Mon Sep 17 00:00:00 2001 From: Chao Sun Date: Tue, 18 Aug 2026 22:30:20 -0700 Subject: [PATCH 3/5] test: cover nested Variant collections in native scans --- .../apache/comet/rules/CometScanRule.scala | 3 ++ .../sql-tests/expressions/misc/variant.sql | 31 +++++++++++++++++++ .../comet/CometIcebergNativeSuite.scala | 29 +++++++++++++---- 3 files changed, 57 insertions(+), 6 deletions(-) diff --git a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala index ebe15a3a606..934a40b8faa 100644 --- a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala +++ b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala @@ -550,6 +550,9 @@ case class CometScanRule(session: SparkSession) try { val fullSchema = IcebergReflection.toSparkSchema(metadata.tableSchema) 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, diff --git a/spark/src/test/resources/sql-tests/expressions/misc/variant.sql b/spark/src/test/resources/sql-tests/expressions/misc/variant.sql index 6b30a2358db..328254b340a 100644 --- a/spark/src/test/resources/sql-tests/expressions/misc/variant.sql +++ b/spark/src/test/resources/sql-tests/expressions/misc/variant.sql @@ -78,6 +78,37 @@ 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, + variants_by_key MAP, + 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) diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala index c3bb25e10e4..fcc5bbb488f 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala @@ -5240,7 +5240,7 @@ class CometIcebergNativeSuite } } - test("projecting a struct containing an unprojected variant still falls back") { + test("projecting nested variant structs, arrays, and maps still falls back") { assume(isSpark40Plus, "VARIANT type requires Spark 4.0+") assume(icebergAvailable, "Iceberg not available in classpath") assume(icebergVersionAtLeast(1, 10), "VARIANT type requires Iceberg 1.10+") @@ -5256,19 +5256,26 @@ class CometIcebergNativeSuite try { spark.sql( s"CREATE TABLE $table " + - "(id BIGINT, nested STRUCT) USING iceberg " + + "(id BIGINT, nested STRUCT, " + + "variants ARRAY, variants_by_key MAP) USING iceberg " + "TBLPROPERTIES ('format-version' = '3')") spark.sql(s""" INSERT INTO $table VALUES - (1, named_struct('label', 'first', 'data', parse_json('{"num": 1}'))), - (2, named_struct('label', NULL, 'data', NULL)), - (3, NULL) + (1, named_struct('label', 'first', 'data', parse_json('{"num": 1}')), + array(parse_json('{"num": 2}'), parse_json('null')), + map('first', parse_json('{"num": 3}'))), + (2, named_struct('label', NULL, 'data', NULL), + array(CAST(NULL AS VARIANT)), map('sql-null', CAST(NULL AS VARIANT))), + (3, NULL, NULL, NULL) """) var sparkScalarRows = Seq.empty[Row] + // Iceberg 1.10 cannot materialize nested Variant values in arrays or maps, but + // reading their sizes still projects every Variant-bearing root for Spark parity. withSQLConf(CometConf.COMET_ENABLED.key -> "false") { sparkScalarRows = spark - .sql(s"SELECT id, nested FROM $table ORDER BY id") + .sql(s"SELECT id, nested, size(variants), size(variants_by_key) " + + s"FROM $table ORDER BY id") .collect() .map(row => Row(row.get(0))) .toSeq @@ -5282,6 +5289,16 @@ class CometIcebergNativeSuite collectIcebergNativeScans(nestedProjection.queryExecution.executedPlan).isEmpty, "iceberg-rust rejects a projected parent containing a VARIANT field") + val arrayProjection = spark.sql(s"SELECT variants FROM $table ORDER BY id") + assert( + collectIcebergNativeScans(arrayProjection.queryExecution.executedPlan).isEmpty, + "iceberg-rust rejects a projected array containing VARIANT values") + + val mapProjection = spark.sql(s"SELECT variants_by_key FROM $table ORDER BY id") + assert( + collectIcebergNativeScans(mapProjection.queryExecution.executedPlan).isEmpty, + "iceberg-rust rejects a projected map containing VARIANT values") + val snapshotId = spark .sql(s"SELECT snapshot_id FROM $table.snapshots ORDER BY committed_at DESC LIMIT 1") .collect() From a7d0539666b979315a986fdb078c2b7eb76898fc Mon Sep 17 00:00:00 2001 From: Chao Sun Date: Wed, 19 Aug 2026 10:37:57 -0700 Subject: [PATCH 4/5] test: cover Variant equality delete scan fallback --- .../comet/CometIcebergNativeSuite.scala | 61 ++++++++++++++++++- 1 file changed, 60 insertions(+), 1 deletion(-) diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala index fcc5bbb488f..dc6fa791e63 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala @@ -5240,6 +5240,65 @@ class CometIcebergNativeSuite } } + test("variant equality deletes fall back to Spark") { + assume(isSpark40Plus, "VARIANT type requires Spark 4.0+") + assume(icebergAvailable, "Iceberg not available in classpath") + assume(icebergVersionAtLeast(1, 10), "VARIANT type requires Iceberg 1.10+") + withTempIcebergDir { warehouseDir => + withSQLConf( + "spark.sql.catalog.test_cat" -> "org.apache.iceberg.spark.SparkCatalog", + "spark.sql.catalog.test_cat.type" -> "hadoop", + "spark.sql.catalog.test_cat.warehouse" -> warehouseDir.getAbsolutePath, + CometConf.COMET_ENABLED.key -> "true", + CometConf.COMET_EXEC_ENABLED.key -> "true", + CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") { + val tableName = "variant_equality_delete" + val table = s"test_cat.db.$tableName" + try { + spark.sql( + s"CREATE TABLE $table (id BIGINT, data VARIANT) USING iceberg " + + "TBLPROPERTIES ('format-version' = '3')") + spark.sql( + s"INSERT INTO $table VALUES " + + "(1, parse_json('1')), (2, parse_json('2'))") + + val nativePlan = spark.sql(s"SELECT id FROM $table ORDER BY id") + assertSingleNativeScan(nativePlan.queryExecution.executedPlan) + + // Variant classes do not exist in the Iceberg versions used by older Spark profiles. + val variantsClass = Class.forName("org.apache.iceberg.variants.Variants") + val metadata = variantsClass.getMethod("emptyMetadata").invoke(null) + val value = variantsClass + .getMethod("of", java.lang.Integer.TYPE) + .invoke(null, Integer.valueOf(2)) + val variant = Class + .forName("org.apache.iceberg.variants.Variant") + .getMethod( + "of", + Class.forName("org.apache.iceberg.variants.VariantMetadata"), + Class.forName("org.apache.iceberg.variants.VariantValue")) + .invoke(null, metadata, value) + + commitEqualityDelete("test_cat", "db", tableName, "data", variant, warehouseDir) + + // Spark also lacks a Variant equality comparator, so verify the fallback plan only. + val fallbackPlan = + spark.sql(s"SELECT id FROM $table ORDER BY id").queryExecution.executedPlan + assert( + collectIcebergNativeScans(fallbackPlan).isEmpty, + "A VARIANT equality-delete key must prevent the native Iceberg scan") + val fallbackReasons = new ExtendedExplainInfo().getFallbackReasons(fallbackPlan) + assert( + fallbackReasons.exists( + _.contains("Equality delete on unsupported column type 'data' (variant)")), + s"Expected VARIANT equality-delete fallback, found: ${fallbackReasons.mkString(", ")}") + } finally { + spark.sql(s"DROP TABLE IF EXISTS $table PURGE") + } + } + } + } + test("projecting nested variant structs, arrays, and maps still falls back") { assume(isSpark40Plus, "VARIANT type requires Spark 4.0+") assume(icebergAvailable, "Iceberg not available in classpath") @@ -5274,7 +5333,7 @@ class CometIcebergNativeSuite // reading their sizes still projects every Variant-bearing root for Spark parity. withSQLConf(CometConf.COMET_ENABLED.key -> "false") { sparkScalarRows = spark - .sql(s"SELECT id, nested, size(variants), size(variants_by_key) " + + .sql("SELECT id, nested, size(variants), size(variants_by_key) " + s"FROM $table ORDER BY id") .collect() .map(row => Row(row.get(0))) From 6b170597ff50d4fb2c0073fb2bd8a8a458b7c65c Mon Sep 17 00:00:00 2001 From: Chao Sun Date: Wed, 19 Aug 2026 13:55:47 -0700 Subject: [PATCH 5/5] test: use IcebergGenerics to read Variant equality-delete key --- .../comet/CometIcebergNativeSuite.scala | 30 +++++++++++-------- 1 file changed, 17 insertions(+), 13 deletions(-) diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala index dc6fa791e63..2e6dd310334 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala @@ -26,6 +26,9 @@ import java.nio.charset.StandardCharsets.UTF_8 import scala.collection.mutable import scala.jdk.CollectionConverters._ +import org.apache.iceberg.data.IcebergGenerics +import org.apache.iceberg.expressions.Expressions +import org.apache.iceberg.spark.Spark3Util import org.apache.spark.CometListenerBusUtils import org.apache.spark.scheduler.{SparkListener, SparkListenerTaskEnd} import org.apache.spark.sql.{CometTestBase, DataFrame, Row} @@ -5265,19 +5268,20 @@ class CometIcebergNativeSuite val nativePlan = spark.sql(s"SELECT id FROM $table ORDER BY id") assertSingleNativeScan(nativePlan.queryExecution.executedPlan) - // Variant classes do not exist in the Iceberg versions used by older Spark profiles. - val variantsClass = Class.forName("org.apache.iceberg.variants.Variants") - val metadata = variantsClass.getMethod("emptyMetadata").invoke(null) - val value = variantsClass - .getMethod("of", java.lang.Integer.TYPE) - .invoke(null, Integer.valueOf(2)) - val variant = Class - .forName("org.apache.iceberg.variants.Variant") - .getMethod( - "of", - Class.forName("org.apache.iceberg.variants.VariantMetadata"), - Class.forName("org.apache.iceberg.variants.VariantValue")) - .invoke(null, metadata, value) + // Reuse the table's existing Variant without referencing newer Iceberg Variant APIs. + val records = IcebergGenerics + .read(Spark3Util.loadIcebergTable(spark, table)) + .where(Expressions.equal("id", 2L)) + .select("data") + .build() + val variant = + try { + val rows = records.iterator() + assert(rows.hasNext, "Expected an Iceberg row containing the delete key") + rows.next().getField("data") + } finally { + records.close() + } commitEqualityDelete("test_cat", "db", tableName, "data", variant, warehouseDir)