You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/10/02 17:26:29 UTC
(datafusion-comet) 01/01: fix: fall back from the native Iceberg scan when a nested field was added or renamed (#6543)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-6543-b7cb18cf0f2f2a16e07939abf6db015cb324b269
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit 2b04fa45374316f9b60ebdcbd147f62e1a1fc527
Author: Andy Grove <ag...@apache.org>
AuthorDate: Fri Oct 2 17:26:06 2026 +0000
fix: fall back from the native Iceberg scan when a nested field was added or renamed (#6543)
iceberg-rust does not match a data file's nested fields to the table's by
field id (apache/iceberg-rust#2617). A file written before a nested field was
added fails the native scan with "Incorrect number of arrays for StructArray
fields", and a nested field renamed since the file was written can read back
as NULL. CometScanRule now declines the native scan when a projected column
has a nested field that some schema in the table's history lacks or names
differently.
Closes #6504.
---
docs/source/user-guide/latest/iceberg.md | 4 +
.../apache/comet/iceberg/IcebergReflection.scala | 68 ++++++
.../org/apache/comet/rules/CometScanRule.scala | 72 ++++--
.../org/apache/comet/CometIcebergNativeSuite.scala | 271 +++++++++++++++++++++
4 files changed, 400 insertions(+), 15 deletions(-)
diff --git a/docs/source/user-guide/latest/iceberg.md b/docs/source/user-guide/latest/iceberg.md
index a0a0b52169..fd3e0f362a 100644
--- a/docs/source/user-guide/latest/iceberg.md
+++ b/docs/source/user-guide/latest/iceberg.md
@@ -192,6 +192,10 @@ The following scenarios will fall back to the JVM Iceberg reader:
- Scans with residual filters using `truncate`, `bucket`, `year`, `month`, `day`, or `hour`
transform functions (partition pruning still works, but row-level filtering of these
transforms falls back)
+- Scans that read a struct, array, or map column with a nested field that schema evolution added
+ or renamed. The native reader cannot yet match such a field to data files written before the
+ change. The check uses the table's schema history, so the fallback stays after those files are
+ rewritten
Writes are not covered by this list. By default Iceberg writes use Spark's own writer; see
[Iceberg Writes](iceberg-writes.md) for the experimental native writer and when it applies.
diff --git a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala
index ecfa392d7d..a96dc952db 100644
--- a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala
+++ b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala
@@ -1149,6 +1149,74 @@ object IcebergReflection extends Logging {
here ++ nested
}
+ /**
+ * Returns the nested fields of the `fieldIds` columns of `schema` that some schema in the
+ * table's history (`table.schemas()`) lacks or names differently. Each is a dotted path with
+ * the change, for example `items.element.z (renamed from a)`. A data file written under that
+ * older schema lacks the field, or has it under the old name. Fields are matched by id at every
+ * level (struct fields, list elements, map keys and values), so a reorder or a type promotion
+ * is not a change, and an older schema that lacks the column itself is skipped. Throws on
+ * reflection failure so the caller can fall back.
+ */
+ def nestedFieldsAddedOrRenamed(table: Any, schema: Any, fieldIds: Set[Int]): Seq[String] = {
+ import scala.jdk.CollectionConverters._
+ val findField = getMethod(schema.getClass, "findField", classOf[Int])
+ def fieldById(s: Any, id: Int): Option[Any] =
+ Option(findField.invoke(s, id.asInstanceOf[AnyRef]))
+ lazy val history = getMethod(table.getClass, "schemas")
+ .invoke(table)
+ .asInstanceOf[java.util.Map[_, _]]
+ .values()
+ .asScala
+ .toSeq
+ fieldIds.toSeq.flatMap { id =>
+ fieldById(schema, id).filter(childFields(_).nonEmpty).toSeq.flatMap { field =>
+ history.flatMap(fieldById(_, id)).flatMap { older =>
+ childFieldsChangedFrom(field, older, fieldName(field))
+ }
+ }
+ }.distinct
+ }
+
+ /** The nested fields under `field`, at any depth, that `older` lacks or names differently. */
+ private def childFieldsChangedFrom(field: Any, older: Any, path: String): Seq[String] = {
+ val olderChildren = childFields(older).map(child => fieldIdOf(child) -> child).toMap
+ childFields(field).flatMap { child =>
+ val name = fieldName(child)
+ val childPath = s"$path.$name"
+ olderChildren.get(fieldIdOf(child)) match {
+ case Some(olderChild) =>
+ val olderName = fieldName(olderChild)
+ val renamed =
+ if (olderName != name) Seq(s"$childPath (renamed from $olderName)") else Nil
+ renamed ++ childFieldsChangedFrom(child, olderChild, childPath)
+ case None => Seq(s"$childPath (added)")
+ }
+ }
+ }
+
+ /** A struct's fields, a list's element, or a map's key and value. Empty for other types. */
+ private def childFields(field: Any): Seq[Any] = {
+ import scala.jdk.CollectionConverters._
+ val fieldType = getMethod(field.getClass, "type").invoke(field)
+ if (getMethod(fieldType.getClass, "isNestedType").invoke(fieldType).asInstanceOf[Boolean]) {
+ val nestedType = getMethod(fieldType.getClass, "asNestedType").invoke(fieldType)
+ getMethod(nestedType.getClass, "fields")
+ .invoke(nestedType)
+ .asInstanceOf[java.util.List[_]]
+ .asScala
+ .toSeq
+ } else {
+ Nil
+ }
+ }
+
+ private def fieldName(field: Any): String =
+ getMethod(field.getClass, "name").invoke(field).asInstanceOf[String]
+
+ private def fieldIdOf(field: Any): Int =
+ getMethod(field.getClass, "fieldId").invoke(field).asInstanceOf[Int]
+
/**
* Converts an Iceberg `Schema` to the Spark `StructType` it reads as, via
* `SparkSchemaUtil.convert`. Comet serializes the whole table/scan schema to native (not just
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 946efa54c0..895df2d566 100644
--- a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
+++ b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
@@ -727,16 +727,27 @@ case class CometScanRule(session: SparkSession)
}
}
+ // Projected roots are matched by field ID so historical snapshots still identify renamed
+ // columns. Both schema checks below use them.
+ val projectedDataColumns = scanExec.output.filterNot(_.isMetadataCol)
+ val resolver = session.sessionState.conf.resolver
+ 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)
+
// 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.
+ // 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)
- 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.
@@ -747,15 +758,7 @@ case class CometScanRule(session: SparkSession)
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 ||
@@ -771,6 +774,44 @@ case class CometScanRule(session: SparkSession)
false
}
+ // Neither iceberg-rust nor the batch adaptation after it matches a data file's nested
+ // fields to the table's by field id (apache/iceberg-rust#2617). A file written before a
+ // nested field was added (to a struct, or to a struct inside a list or map) fails the
+ // native scan with "Incorrect number of arrays for StructArray fields", and a nested field
+ // renamed since the file was written can read back as NULL. The native read asks for a
+ // projected column's full nested type even when Spark prunes it, taken from the current
+ // table schema, or from the scan schema when VERSION AS OF reads a dropped column (see
+ // CometIcebergNativeScan). A FileScanTask does not record which schema wrote its file, so
+ // fall back when any schema in the table's history lacks a nested field of a projected
+ // column or names it differently.
+ val nestedFieldsSupported =
+ try {
+ val schemas = Seq(metadata.tableSchema, metadata.scanSchema)
+ val fieldIds =
+ if (hasUnresolvedProjectedFieldIds) {
+ schemas.flatMap(IcebergReflection.buildFieldIdMapping(_).values).toSet
+ } else {
+ resolvedProjectedFieldIds
+ }
+ val changed = schemas
+ .flatMap { schema =>
+ IcebergReflection.nestedFieldsAddedOrRenamed(metadata.table, schema, fieldIds)
+ }
+ .distinct
+ .sorted
+ if (changed.nonEmpty) {
+ fallbackReasons += "Nested fields added or renamed by Iceberg schema evolution are " +
+ "not yet supported by Comet's native reader, which cannot match them to data " +
+ s"files written before the change: ${changed.mkString(", ")}"
+ }
+ changed.isEmpty
+ } catch {
+ case e: Exception =>
+ fallbackReasons += "Iceberg reflection failure: could not compare nested fields " +
+ s"with the table's schema history: ${e.getMessage}"
+ false
+ }
+
// The native Parquet reader decrypts 128-bit and 256-bit AES-GCM data keys (16- or
// 32-byte). 192-bit is unsupported because the underlying crypto has no AES-192-GCM. Fall
// back for anything else. None means the table is unencrypted. Reflection failure also
@@ -1006,7 +1047,8 @@ case class CometScanRule(session: SparkSession)
defaultValuesSupported && schemaTypesSupported && encryptionKeyLengthSupported &&
taskValidation.allParquet && allSupportedFilesystems && allLocationsOpenable &&
metadataSchemeSupported && partitionTypesSupported && unifiedPartitionTypeSupported &&
- transformFunctionsSupported && deleteFileTypesSupported && dppSubqueriesSupported) {
+ transformFunctionsSupported && deleteFileTypesSupported && dppSubqueriesSupported &&
+ nestedFieldsSupported) {
CometBatchScanExec(
scanExec.clone().asInstanceOf[BatchScanExec],
runtimeFilters = scanExec.runtimeFilters,
diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala
index 531bb77894..57291131c7 100644
--- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala
@@ -102,6 +102,19 @@ class CometIcebergNativeSuite
s"Plan:\n$cometPlan")
}
+ /**
+ * Verifies query correctness and that it falls back to Spark because a projected column's
+ * nested fields were added or renamed over the table's schema history, as `fields` lists them
+ * (for example `s.b (added)`).
+ */
+ private def checkNestedFieldEvolutionFallback(query: String, fields: String): Unit = {
+ val (_, cometPlan) =
+ checkSparkAnswerAndFallbackReason(query, s"files written before the change: $fields")
+ assert(
+ collectIcebergNativeScans(cometPlan).isEmpty,
+ s"Expected fallback to Spark but found a CometIcebergNativeScanExec. Plan:\n$cometPlan")
+ }
+
/** Counts non-overlapping occurrences of `needle` within `haystack`. */
private def countByteOccurrences(haystack: Array[Byte], needle: Array[Byte]): Int = {
require(needle.nonEmpty, "needle must be non-empty")
@@ -1753,6 +1766,264 @@ class CometIcebergNativeSuite
}
}
+ // The native read does not match a data file's nested fields to the table's by field id
+ // (apache/iceberg-rust#2617). It fails on a file that lacks a nested field the table has gained,
+ // and can return NULL for a nested field renamed since the file was written, so these reads
+ // must fall back.
+ test("schema evolution - nested field added to a struct falls back") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+
+ 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.evo_struct"
+ spark.sql(s"CREATE TABLE $table (id INT, s STRUCT<a: INT>) USING iceberg")
+ spark.sql(s"INSERT INTO $table VALUES (1, named_struct('a', 1)), (2, null)")
+ val snapshotBeforeAdd = spark
+ .sql(s"SELECT snapshot_id FROM $table.snapshots ORDER BY committed_at DESC LIMIT 1")
+ .collect()(0)
+ .getLong(0)
+ spark.sql(s"ALTER TABLE $table ADD COLUMN s.b INT")
+ spark.sql(s"INSERT INTO $table VALUES (3, named_struct('a', 3, 'b', 30))")
+
+ checkNestedFieldEvolutionFallback(s"SELECT id, s FROM $table ORDER BY id", "s.b (added)")
+ // Spark pushes these null checks into the scan.
+ checkNestedFieldEvolutionFallback(
+ s"SELECT id, s FROM $table WHERE s IS NOT NULL ORDER BY id",
+ "s.b (added)")
+ checkNestedFieldEvolutionFallback(
+ s"SELECT id FROM $table WHERE s IS NULL ORDER BY id",
+ "s.b (added)")
+ // The native read asks for the column's full nested type even when Spark prunes it.
+ checkNestedFieldEvolutionFallback(
+ s"SELECT id, s.a FROM $table ORDER BY id",
+ "s.b (added)")
+ checkNestedFieldEvolutionFallback(
+ s"SELECT l.id, r.id FROM $table l JOIN $table r ON l.s = r.s ORDER BY l.id",
+ "s.b (added)")
+ // The old snapshot's files lack s.b too, and the native read uses the current schema.
+ checkNestedFieldEvolutionFallback(
+ s"SELECT id, s FROM $table VERSION AS OF $snapshotBeforeAdd ORDER BY id",
+ "s.b (added)")
+
+ // A read that does not project the evolved column stays native.
+ checkIcebergNativeScan(s"SELECT id FROM $table ORDER BY id")
+
+ spark.sql(s"DROP TABLE $table")
+ }
+ }
+ }
+
+ test("schema evolution - nested field added inside an array or a map falls back") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+
+ 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 listTable = "test_cat.db.evo_list"
+ spark.sql(s"CREATE TABLE $listTable (id INT, items ARRAY<STRUCT<a: INT>>) USING iceberg")
+ spark.sql(s"INSERT INTO $listTable VALUES (1, array(named_struct('a', 10))), (2, null)")
+ spark.sql(s"ALTER TABLE $listTable ADD COLUMN items.element.b INT")
+ spark.sql(s"INSERT INTO $listTable VALUES (3, array(named_struct('a', 30, 'b', 300)))")
+
+ checkNestedFieldEvolutionFallback(
+ s"SELECT id, items FROM $listTable ORDER BY id",
+ "items.element.b (added)")
+ // Spark infers isnotnull(items) below a non-outer generator and pushes it into the scan.
+ checkNestedFieldEvolutionFallback(
+ s"SELECT id, e.a, e.b FROM $listTable LATERAL VIEW explode(items) x AS e ORDER BY id",
+ "items.element.b (added)")
+ checkNestedFieldEvolutionFallback(
+ s"SELECT id, posexplode(items) FROM $listTable ORDER BY id",
+ "items.element.b (added)")
+ checkNestedFieldEvolutionFallback(
+ s"SELECT id, inline(items) FROM $listTable ORDER BY id",
+ "items.element.b (added)")
+
+ val mapTable = "test_cat.db.evo_map"
+ spark.sql(s"CREATE TABLE $mapTable (id INT, m MAP<STRING, STRUCT<a: INT>>) USING iceberg")
+ spark.sql(s"INSERT INTO $mapTable VALUES (1, map('k', named_struct('a', 1))), (2, null)")
+ spark.sql(s"ALTER TABLE $mapTable ADD COLUMN m.value.b INT")
+ spark.sql(s"INSERT INTO $mapTable VALUES (3, map('k', named_struct('a', 3, 'b', 30)))")
+
+ checkNestedFieldEvolutionFallback(
+ s"SELECT id, m FROM $mapTable ORDER BY id",
+ "m.value.b (added)")
+ checkNestedFieldEvolutionFallback(
+ s"SELECT id, explode(m) FROM $mapTable ORDER BY id",
+ "m.value.b (added)")
+
+ // A field added several levels down, and a struct-typed field, are found too.
+ val deepTable = "test_cat.db.evo_deep"
+ spark.sql(
+ s"CREATE TABLE $deepTable (id INT, s STRUCT<l: ARRAY<STRUCT<a: INT>>>) USING iceberg")
+ spark.sql(
+ s"INSERT INTO $deepTable VALUES (1, named_struct('l', array(named_struct('a', 1))))")
+ spark.sql(s"ALTER TABLE $deepTable ADD COLUMN s.l.element.b INT")
+ spark.sql(s"ALTER TABLE $deepTable ADD COLUMN s.t STRUCT<x: INT>")
+ checkNestedFieldEvolutionFallback(
+ s"SELECT id, s FROM $deepTable ORDER BY id",
+ "s.l.element.b (added), s.t (added)")
+
+ spark.sql(s"DROP TABLE $listTable")
+ spark.sql(s"DROP TABLE $mapTable")
+ spark.sql(s"DROP TABLE $deepTable")
+ }
+ }
+ }
+
+ test("schema evolution - nested field dropped and re-added under the same name falls back") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+
+ 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") {
+
+ // The re-added s.b has a new field id, so the old file's s.b is not it. Drop and re-add it
+ // in one schema change, so every schema in the history has a field named s.b and only
+ // the field id tells them apart.
+ val table = "test_cat.db.evo_readd"
+ spark.sql(s"CREATE TABLE $table (id INT, s STRUCT<a: INT, b: INT>) USING iceberg")
+ spark.sql(s"INSERT INTO $table VALUES (1, named_struct('a', 1, 'b', 2))")
+ loadIcebergTable(spark, "test_cat", "db", "evo_readd")
+ .asInstanceOf[org.apache.iceberg.Table]
+ .updateSchema()
+ .deleteColumn("s.b")
+ .addColumn("s", "b", org.apache.iceberg.types.Types.IntegerType.get())
+ .commit()
+ spark.sql(s"REFRESH TABLE $table")
+ spark.sql(s"INSERT INTO $table VALUES (2, named_struct('a', 3, 'b', 4))")
+
+ checkNestedFieldEvolutionFallback(s"SELECT id, s FROM $table ORDER BY id", "s.b (added)")
+
+ spark.sql(s"DROP TABLE $table")
+ }
+ }
+ }
+
+ test("schema evolution - nested field renamed falls back") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+
+ 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.evo_rename"
+ spark.sql(s"""
+ CREATE TABLE $table (
+ id INT,
+ s STRUCT<a: INT, b: STRING>,
+ items ARRAY<STRUCT<a: INT, b: INT>>,
+ moved STRUCT<a: INT, b: INT>
+ ) USING iceberg
+ """)
+ // The nulls make Iceberg write the nested fields as optional, as the table declares them,
+ // which is the layout where the native read returns NULL for a renamed field.
+ spark.sql(s"""
+ INSERT INTO $table VALUES
+ (1, named_struct('a', 1, 'b', 'x'), array(named_struct('a', 1, 'b', 2)),
+ named_struct('a', 1, 'b', 2)),
+ (2, named_struct('a', CAST(NULL AS INT), 'b', CAST(NULL AS STRING)),
+ array(CAST(NULL AS STRUCT<a: INT, b: INT>)),
+ named_struct('a', CAST(NULL AS INT), 'b', CAST(NULL AS INT)))
+ """)
+ spark.sql(s"ALTER TABLE $table RENAME COLUMN s.a TO z")
+ spark.sql(s"ALTER TABLE $table RENAME COLUMN items.element.a TO z")
+ spark.sql(s"ALTER TABLE $table ALTER COLUMN moved.b FIRST")
+ spark.sql(s"ALTER TABLE $table RENAME COLUMN moved.a TO c")
+ spark.sql(s"""
+ INSERT INTO $table VALUES
+ (3, named_struct('z', 3, 'b', 'y'), array(named_struct('z', 3, 'b', 4)),
+ named_struct('b', 4, 'c', 3))
+ """)
+
+ checkNestedFieldEvolutionFallback(
+ s"SELECT id, s FROM $table ORDER BY id",
+ "s.z (renamed from a)")
+ checkNestedFieldEvolutionFallback(
+ s"SELECT id, items FROM $table ORDER BY id",
+ "items.element.z (renamed from a)")
+ checkNestedFieldEvolutionFallback(
+ s"SELECT id, moved FROM $table ORDER BY id",
+ "moved.c (renamed from a)")
+
+ spark.sql(s"DROP TABLE $table")
+ }
+ }
+ }
+
+ test("schema evolution - nested field dropped, moved, or promoted stays native") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+
+ 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.evo_no_add"
+ spark.sql(s"""
+ CREATE TABLE $table (
+ id INT,
+ dropped STRUCT<a: INT, b: INT>,
+ moved STRUCT<a: INT, b: INT>,
+ promoted STRUCT<a: INT>
+ ) USING iceberg
+ """)
+ // Spark 4 writes the first file's nested fields as required and the second's as optional.
+ // The native read casts the first layout and passes the second through, so cover both.
+ spark.sql(s"""
+ INSERT INTO $table VALUES
+ (1, named_struct('a', 1, 'b', 2), named_struct('a', 1, 'b', 2), named_struct('a', 1))
+ """)
+ spark.sql(s"""
+ INSERT INTO $table VALUES
+ (2, named_struct('a', 2, 'b', 20), named_struct('a', 2, 'b', 20), named_struct('a', 2)),
+ (3, named_struct('a', CAST(NULL AS INT), 'b', CAST(NULL AS INT)),
+ named_struct('a', CAST(NULL AS INT), 'b', CAST(NULL AS INT)),
+ named_struct('a', CAST(NULL AS INT)))
+ """)
+ spark.sql(s"ALTER TABLE $table DROP COLUMN dropped.b")
+ spark.sql(s"ALTER TABLE $table ALTER COLUMN moved.b FIRST")
+ spark.sql(s"ALTER TABLE $table ALTER COLUMN promoted.a TYPE BIGINT")
+ spark.sql(s"""
+ INSERT INTO $table VALUES
+ (4, named_struct('a', 4), named_struct('b', 40, 'a', 4),
+ named_struct('a', 3000000000L))
+ """)
+
+ checkIcebergNativeScan(s"SELECT * FROM $table ORDER BY id")
+
+ spark.sql(s"DROP TABLE $table")
+ }
+ }
+ }
+
test("migration - basic read after migration (fallback for no field ID)") {
assume(icebergAvailable, "Iceberg not available in classpath")
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org