diff --git a/docs/source/user-guide/latest/iceberg.md b/docs/source/user-guide/latest/iceberg.md index fd333043013..13f811a68c6 100644 --- a/docs/source/user-guide/latest/iceberg.md +++ b/docs/source/user-guide/latest/iceberg.md @@ -217,10 +217,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 +- Scans that read a struct column, or a struct inside an array or map, whose fields were all added + by schema evolution after some data file was written. For such a file the native reader returns a + struct of NULLs or NULL, depending on what else the scan reads, where Spark does not. The check + uses the table's schema history, so the fallback stays after those files are rewritten - Scans that read a struct, array, or map column from a table with `schema.name-mapping.default`, including tables imported with `snapshot`, `migrate`, or `add_files` ([#6790](https://github.com/apache/datafusion-comet/issues/6790)). The native reader @@ -801,7 +801,8 @@ Which entries apply depends on which native paths are enabled: ### Reads that differ from Spark - **Data files with neither field ids nor a name mapping.** Spark reads every nested field of such - a file as NULL, and the native scan returns the values stored in the file + a file as NULL, and the native scan returns the values stored in the file. A read of only some + of a struct's fields fails instead ([#6794](https://github.com/apache/datafusion-comet/issues/6794)). - **`VERSION AS OF` reads of a column renamed after the snapshot.** The renamed column reads as NULL, and two columns that swapped names return each other's values diff --git a/native/Cargo.lock b/native/Cargo.lock index 80741b85406..0c4ba356c32 100644 --- a/native/Cargo.lock +++ b/native/Cargo.lock @@ -3568,7 +3568,7 @@ dependencies = [ [[package]] name = "iceberg" version = "0.10.1" -source = "git+https://github.com/apache/iceberg-rust?rev=1f3bc34058e3f3e45336bc5428f37b92649a608b#1f3bc34058e3f3e45336bc5428f37b92649a608b" +source = "git+https://github.com/apache/iceberg-rust?rev=b154603a710d730d2f43f8af4db257cc137237ef#b154603a710d730d2f43f8af4db257cc137237ef" dependencies = [ "aes-gcm", "anyhow", @@ -3626,7 +3626,7 @@ dependencies = [ [[package]] name = "iceberg-property-macro" version = "0.10.1" -source = "git+https://github.com/apache/iceberg-rust?rev=1f3bc34058e3f3e45336bc5428f37b92649a608b#1f3bc34058e3f3e45336bc5428f37b92649a608b" +source = "git+https://github.com/apache/iceberg-rust?rev=b154603a710d730d2f43f8af4db257cc137237ef#b154603a710d730d2f43f8af4db257cc137237ef" dependencies = [ "proc-macro2", "quote", @@ -3636,7 +3636,7 @@ dependencies = [ [[package]] name = "iceberg-storage-opendal" version = "0.10.1" -source = "git+https://github.com/apache/iceberg-rust?rev=1f3bc34058e3f3e45336bc5428f37b92649a608b#1f3bc34058e3f3e45336bc5428f37b92649a608b" +source = "git+https://github.com/apache/iceberg-rust?rev=b154603a710d730d2f43f8af4db257cc137237ef#b154603a710d730d2f43f8af4db257cc137237ef" dependencies = [ "anyhow", "async-trait", diff --git a/native/Cargo.toml b/native/Cargo.toml index c858f9e715b..6f3b771e8ee 100644 --- a/native/Cargo.toml +++ b/native/Cargo.toml @@ -85,8 +85,8 @@ aws-sdk-sts = { version = "1.114.0", default-features = false } # Pinned to a revision rather than a release because no published iceberg-rust # version is on our arrow: 0.10.1 requires arrow/parquet ^58. Move to a plain # version requirement once a release ships on arrow 59. Bump policy: #5645. -iceberg = { git = "https://github.com/apache/iceberg-rust", rev = "1f3bc34058e3f3e45336bc5428f37b92649a608b" } -iceberg-storage-opendal = { git = "https://github.com/apache/iceberg-rust", rev = "1f3bc34058e3f3e45336bc5428f37b92649a608b", features = ["opendal-memory", "opendal-fs", "opendal-s3", "opendal-gcs", "opendal-oss", "opendal-azdls"] } +iceberg = { git = "https://github.com/apache/iceberg-rust", rev = "b154603a710d730d2f43f8af4db257cc137237ef" } +iceberg-storage-opendal = { git = "https://github.com/apache/iceberg-rust", rev = "b154603a710d730d2f43f8af4db257cc137237ef", features = ["opendal-memory", "opendal-fs", "opendal-s3", "opendal-gcs", "opendal-oss", "opendal-azdls"] } reqsign-core = "3" [profile.release] diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs index 0241e1f1ce8..c6641943e8f 100644 --- a/native/core/src/execution/planner.rs +++ b/native/core/src/execution/planner.rs @@ -2739,7 +2739,11 @@ fn parse_file_scan_tasks_from_common( .with_partition_spec(partition_spec) .with_name_mapping(name_mapping) .with_unified_partition_type(unified_partition_type_for_task) - .with_case_sensitive(false) + // The only consumer is the binding of an equality delete's predicate, whose + // column names iceberg-rust takes from the task schema. They match it exactly, + // and a case-insensitive binding rejects a schema with two names that differ + // only by case. + .with_case_sensitive(true) // Plaintext StandardKeyMetadata forwarded verbatim from the JVM; decoded by // iceberg-rust with no KMS unwrap. None for unencrypted data files. .with_key_metadata(proto_task.key_metadata.clone().map(Vec::into_boxed_slice)) 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 bd30f12496e..ff318e0dfb6 100644 --- a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala +++ b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala @@ -1283,6 +1283,24 @@ object IcebergReflection extends Logging { columns.asScala.flatMap(walkFieldForDefault).toList } + /** + * [[columnsWithInitialDefault]] for only the top-level columns of `schema` whose ids are in + * `columnIds`. A scan reads every nested field of such a column that its schema holds, so this + * is for the schema the scan reads when that is wider than the query's, as the table's is while + * nested schema pruning is off. + */ + def columnsWithInitialDefault(schema: Any, columnIds: Set[Int]): List[String] = { + import scala.jdk.CollectionConverters._ + val columns = + getMethod(schema.getClass, "columns").invoke(schema).asInstanceOf[java.util.List[_]] + columns.asScala + .filter(column => columnIds.contains(fieldIdOf(column))) + .flatMap { + walkFieldForDefault + } + .toList + } + private def walkFieldForDefault(field: Any): List[String] = { import scala.jdk.CollectionConverters._ val name = getMethod(field.getClass, "name").invoke(field).asInstanceOf[String] @@ -1304,15 +1322,15 @@ object IcebergReflection extends Logging { } /** - * 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 + * Returns the structs under the `fieldIds` columns of `schema`, at any depth and including the + * columns themselves, that share none of their fields with the same struct in some schema in + * the table's history (`table.schemas()`). Each is a dotted path, for example + * `items.element.s`. A data file written under that older schema has none of the struct's + * current fields. Structs are matched by field id at every level (struct fields, list elements, + * map keys and values), so 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] = { + def structsWithNoFieldsInCommon(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] = @@ -1324,31 +1342,33 @@ object IcebergReflection extends Logging { .asScala .toSeq fieldIds.toSeq.flatMap { id => - fieldById(schema, id).filter(childFields(_).nonEmpty).toSeq.flatMap { field => + fieldById(schema, id).toSeq.flatMap { field => history.flatMap(fieldById(_, id)).flatMap { older => - childFieldsChangedFrom(field, older, fieldName(field)) + structsWithNoFieldsInCommon(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] = { + /** `field` if it is a struct sharing no field with `older`, plus the same for its children. */ + private def structsWithNoFieldsInCommon(field: Any, older: Any, path: String): Seq[String] = { + val children = childFields(field) 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)") + val disjoint = isStruct(field) && children.nonEmpty && + !children.exists(child => olderChildren.contains(fieldIdOf(child))) + val here = if (disjoint) Seq(path) else Nil + here ++ children.flatMap { child => + olderChildren.get(fieldIdOf(child)).toSeq.flatMap { olderChild => + structsWithNoFieldsInCommon(child, olderChild, s"$path.${fieldName(child)}") } } } + private def isStruct(field: Any): Boolean = { + val fieldType = getMethod(field.getClass, "type").invoke(field) + getMethod(fieldType.getClass, "isStructType").invoke(fieldType).asInstanceOf[Boolean] + } + /** 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._ 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 349a9dae380..955db40ce42 100644 --- a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala +++ b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala @@ -743,17 +743,41 @@ case class CometScanRule(session: SparkSession) false } + // 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) + // Column default values are a V3 feature, so V1/V2 tables cannot have them and need no // check. Only inspect V3 tables. iceberg-rust does not synthesize a V3 initial-default for - // a column absent from a data file, so a scan projecting such a column must fall back. A V3 - // table implies an Iceberg version new enough to expose initialDefault(), so a reflection - // failure here is unexpected and also falls back rather than risk a crash. + // a column absent from a data file (it fails a nested one), so a scan reading such a + // column must fall back. That is the query's schema and, while nested schema pruning is + // off, the whole of each projected column in the task schema, which holds nested fields + // the query's does not. A V3 table implies an Iceberg version new enough to expose + // initialDefault(), so a reflection failure here is unexpected and also falls back rather + // than risk a crash. val defaultValuesSupported = if (!formatVersion.exists(_ >= 3)) { true } else { try { - val defaulted = IcebergReflection.columnsWithInitialDefault(metadata.scanSchema) + val readDefaults = metadata.taskSchema.toSeq.flatMap { schema => + if (hasUnresolvedProjectedFieldIds) { + IcebergReflection.columnsWithInitialDefault(schema) + } else { + IcebergReflection.columnsWithInitialDefault(schema, resolvedProjectedFieldIds) + } + } + val defaulted = + (IcebergReflection.columnsWithInitialDefault( + metadata.scanSchema) ++ readDefaults).distinct if (defaulted.nonEmpty) { fallbackReasons += "Iceberg column(s) with V3 default values are not yet " + s"supported by Comet's native reader: ${defaulted.mkString(", ")}" @@ -769,18 +793,6 @@ 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) - // iceberg-rust applies name mappings only at the root (#6790). Nested columns can // disappear during projection, or reach the adapter with physical field names that // cannot distinguish a mapped alias from a genuinely missing field. Schema history @@ -831,16 +843,17 @@ 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. This check walks the schema - // the scan reads (see CometIcebergNativeScanMetadata.taskSchema), which for VERSION AS OF - // can be older than the table's, and also the table and scan schemas, so it can fall back - // when the changed field is pruned away. 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. + // iceberg-rust matches a data file's nested fields to the table's by field id, filling in + // a field the file lacks with NULLs and applying renames, so schema evolution that adds, + // renames, drops or reorders nested fields reads correctly. One shape still differs from + // Spark: a struct (at any depth) whose fields in the schema the scan reads were all added + // after a file was written. The native read returns a struct of NULLs or NULL there, + // depending on what else the scan reads, where Spark returns NULL for a top-level struct + // and a struct of NULLs for a nested one. The schema the scan reads is the task schema (see + // CometIcebergNativeScanMetadata.taskSchema), which holds only the nested fields Spark + // keeps while nested schema pruning is on, and which for VERSION AS OF can be older than + // the table's. A FileScanTask does not record which schema wrote its file, so fall back + // when any schema in the table's history has such a struct in a projected column. val nestedFieldsSupported = try { val schemas = Seq(metadata.tableSchema, metadata.scanSchema) ++ metadata.taskSchema @@ -850,18 +863,17 @@ case class CometScanRule(session: SparkSession) } else { resolvedProjectedFieldIds } - val changed = schemas - .flatMap { schema => - IcebergReflection.nestedFieldsAddedOrRenamed(metadata.table, schema, fieldIds) - } + val disjoint = metadata.taskSchema.toSeq + .flatMap(IcebergReflection.structsWithNoFieldsInCommon(metadata.table, _, 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(", ")}" + if (disjoint.nonEmpty) { + fallbackReasons += "Structs whose fields were all added by Iceberg schema " + + "evolution are not yet supported by Comet's native reader, which reads data " + + "files written before the change differently from Spark: " + + disjoint.mkString(", ") } - changed.isEmpty + disjoint.isEmpty } catch { case e: Exception => fallbackReasons += "Iceberg reflection failure: could not compare nested fields " + diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala index 5dfbd19fcd0..b8835212129 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala @@ -30,7 +30,7 @@ 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.{CometListenerBusUtils, SparkException} import org.apache.spark.scheduler.{SparkListener, SparkListenerTaskEnd} import org.apache.spark.sql.{CometTestBase, DataFrame, Row} import org.apache.spark.sql.catalyst.expressions.{Ascending, DynamicPruningExpression, SortOrder} @@ -121,13 +121,24 @@ class CometIcebergNativeSuite } /** - * 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)`). + * Verifies query correctness and that it reads `scans` native Iceberg scans, for a query that + * reads the same table more than once. */ - private def checkNestedFieldEvolutionFallback(query: String, fields: String): Unit = { + private def checkIcebergNativeScans(query: String, scans: Int): Unit = { + val (_, cometPlan) = checkSparkAnswer(query) + assert( + collectIcebergNativeScans(cometPlan).length == scans, + s"Expected $scans CometIcebergNativeScanExec nodes. Plan:\n$cometPlan") + } + + /** + * Verifies query correctness and that it falls back to Spark because a projected column has + * structs, listed in `structs` by path (for example `items.element`), whose fields were all + * added after some data files were written. + */ + private def checkNestedFieldEvolutionFallback(query: String, structs: String): Unit = { val (_, cometPlan) = - checkSparkAnswerAndFallbackReason(query, s"files written before the change: $fields") + checkSparkAnswerAndFallbackReason(query, s"differently from Spark: $structs") assert( collectIcebergNativeScans(cometPlan).isEmpty, s"Expected fallback to Spark but found a CometIcebergNativeScanExec. Plan:\n$cometPlan") @@ -1213,6 +1224,80 @@ class CometIcebergNativeSuite } } + // iceberg-rust fails to read a nested field that has a default value and that a data file lacks. + // While nested schema pruning is off, the scan reads every nested field of a projected column, + // including one the query prunes away, so the check on defaults must cover the task schema. + test("nested field with a default value falls back when the scan reads it") { + assume(icebergAvailable, "Iceberg not available in classpath") + assume(icebergVersionAtLeast(1, 11), "Nested column defaults require Iceberg 1.11+") + + 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_default" + spark.sql( + s"CREATE TABLE $table (id INT, s STRUCT) USING iceberg " + + "TBLPROPERTIES ('format-version' = '3')") + spark.sql( + s"INSERT INTO $table VALUES (1, named_struct('a', 1)), (2, named_struct('a', 2))") + addIcebergIntColumnWithDefault( + loadIcebergTable(spark, "test_cat", "db", "nested_default"), + "b", + 7, + parent = Some("s")) + spark.sql(s"REFRESH TABLE $table") + + // The query prunes s.b, so the pruned task schema omits it. + checkIcebergNativeScan(s"SELECT id, s.a FROM $table ORDER BY id") + // With pruning off the scan reads s.b too, and the file lacks it. + withSQLConf(CometConf.COMET_ICEBERG_NESTED_SCHEMA_PRUNING_ENABLED.key -> "false") { + checkIcebergNativeScanFallback( + s"SELECT id, s.a FROM $table ORDER BY id", + "nested default read by the unpruned task schema") + } + checkIcebergNativeScanFallback( + s"SELECT id, s.b FROM $table ORDER BY id", + "nested default read by the query") + spark.sql(s"DROP TABLE $table") + } + } + } + + // iceberg-rust binds an equality delete's key by name, and rejects a case-insensitive name that + // matches two columns. The delete's key is taken from the task schema, spelled exactly as in it, + // so the binding is case sensitive. The query reads neither column, as the batch adaptation + // after iceberg-rust rejects a read of two columns that differ only by case. + test("equality delete on a column whose name differs from another only by case") { + 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, + SQLConf.CASE_SENSITIVE.key -> "true", + CometConf.COMET_ENABLED.key -> "true", + CometConf.COMET_EXEC_ENABLED.key -> "true", + CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") { + val table = "test_cat.db.eq_delete_case" + spark.sql( + s"CREATE TABLE $table (id INT, ID STRING, v STRING) USING iceberg " + + "TBLPROPERTIES ('format-version' = '2')") + spark.sql(s"INSERT INTO $table VALUES (1, 'a', 'x'), (2, 'b', 'y'), (3, 'c', 'z')") + commitEqualityDelete("test_cat", "db", "eq_delete_case", "id", 2, warehouseDir) + + // The pruned task schema holds only `v` and the key `id`. The full one holds both columns. + checkPrunedAndFullReads(s"SELECT v FROM $table ORDER BY v") + spark.sql(s"DROP TABLE $table") + } + } + } + // The key `s.k` is nested, which iceberg-rust cannot look up, so the scan must fall back. test("equality delete keyed on a nested field falls back to Spark") { assume(icebergAvailable, "Iceberg not available in classpath") @@ -1861,11 +1946,7 @@ 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") { + test("schema evolution - nested field added to a struct stays native") { assume(icebergAvailable, "Iceberg not available in classpath") withTempIcebergDir { warehouseDir => @@ -1887,34 +1968,27 @@ class CometIcebergNativeSuite 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)") + // The first file lacks s.b, which the native read fills with NULLs. + checkIcebergNativeScan(s"SELECT id, s FROM $table ORDER BY id") + checkIcebergNativeScan(s"SELECT s FROM $table ORDER BY s.a") // 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 only for s.a here, but the check also covers the column's full - // type in the table schema, so it still falls back. - checkNestedFieldEvolutionFallback( - s"SELECT id, s.a FROM $table ORDER BY id", - "s.b (added)") - checkNestedFieldEvolutionFallback( + checkIcebergNativeScan(s"SELECT id, s FROM $table WHERE s IS NOT NULL ORDER BY id") + checkIcebergNativeScan(s"SELECT id FROM $table WHERE s IS NULL ORDER BY id") + checkIcebergNativeScan(s"SELECT id, s.a FROM $table ORDER BY id") + checkIcebergNativeScans( 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 scan's full schema is the current one, - // which has it. - 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. + 2) + // The scan reads only s.b, which the first file lacks, so that file has none of the fields + // the scan reads. + checkNestedFieldEvolutionFallback(s"SELECT id, s.b FROM $table ORDER BY id", "s") + // The old snapshot's files lack s.b too, and its schema does not have the field. + checkIcebergNativeScan(s"SELECT id, s FROM $table VERSION AS OF $snapshotBeforeAdd") checkIcebergNativeScan(s"SELECT id FROM $table ORDER BY id") // A VERSION AS OF read whose full schema is older than the table's: the pruned-away // nested partition source `s.k` makes the scan read its full schema, and the dropped `c` - // makes that the snapshot's, which still has the s.b that the first file predates. + // makes that the snapshot's, which still has the s.b that the first file predates. The + // native read fills it with NULLs. val historical = "test_cat.db.evo_struct_historical" spark.sql(s""" CREATE TABLE $historical (id INT, c INT, s STRUCT) @@ -1928,12 +2002,11 @@ class CometIcebergNativeSuite val snapshotWithB = latestSnapshotId(historical) spark.sql(s"ALTER TABLE $historical DROP COLUMN c") spark.sql(s"ALTER TABLE $historical DROP COLUMN s.b") - checkNestedFieldEvolutionFallback( - s"SELECT id, c, s.a FROM $historical VERSION AS OF $snapshotWithB ORDER BY id", - "s.b (added)") + checkIcebergNativeScan( + s"SELECT id, c, s.a FROM $historical VERSION AS OF $snapshotWithB ORDER BY id") - // Unpartitioned, the same read takes the pruned schema, which leaves s.b out, so it stays - // native. With pruning off it reads the full schema and falls back. + // Unpartitioned, the same read takes the pruned schema, which leaves s.b out. With pruning + // off it reads the full schema. Both stay native. val unpartitioned = "test_cat.db.evo_struct_historical_unpartitioned" spark.sql(s"CREATE TABLE $unpartitioned (id INT, c INT, s STRUCT) USING iceberg") spark.sql(s"INSERT INTO $unpartitioned VALUES (1, 100, named_struct('a', 10))") @@ -1946,7 +2019,7 @@ class CometIcebergNativeSuite s"SELECT id, c, s.a FROM $unpartitioned VERSION AS OF $unpartitionedSnapshot ORDER BY id" checkIcebergNativeScan(unpartitionedQuery) withSQLConf(CometConf.COMET_ICEBERG_NESTED_SCHEMA_PRUNING_ENABLED.key -> "false") { - checkNestedFieldEvolutionFallback(unpartitionedQuery, "s.b (added)") + checkIcebergNativeScan(unpartitionedQuery) } // The task schema is chosen during planning, so a query planned with pruning on still // reads the pruned schema when it runs with pruning off. @@ -1961,7 +2034,7 @@ class CometIcebergNativeSuite } } - test("schema evolution - nested field added inside an array or a map falls back") { + test("schema evolution - nested field added inside an array or a map stays native") { assume(icebergAvailable, "Iceberg not available in classpath") withTempIcebergDir { warehouseDir => @@ -1979,19 +2052,12 @@ class CometIcebergNativeSuite 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)") + checkIcebergNativeScan(s"SELECT id, items FROM $listTable ORDER BY id") // 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)") + checkIcebergNativeScan( + s"SELECT id, e.a, e.b FROM $listTable LATERAL VIEW explode(items) x AS e ORDER BY id") + checkIcebergNativeScan(s"SELECT id, posexplode(items) FROM $listTable ORDER BY id") + checkIcebergNativeScan(s"SELECT id, inline(items) FROM $listTable ORDER BY id") val mapTable = "test_cat.db.evo_map" spark.sql(s"CREATE TABLE $mapTable (id INT, m MAP>) USING iceberg") @@ -1999,14 +2065,10 @@ class CometIcebergNativeSuite 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)") + checkIcebergNativeScan(s"SELECT id, m FROM $mapTable ORDER BY id") + checkIcebergNativeScan(s"SELECT id, explode(m) FROM $mapTable ORDER BY id") - // A field added several levels down, and a struct-typed field, are found too. + // A field added several levels down, and a struct-typed field. val deepTable = "test_cat.db.evo_deep" spark.sql( s"CREATE TABLE $deepTable (id INT, s STRUCT>>) USING iceberg") @@ -2014,9 +2076,9 @@ class CometIcebergNativeSuite 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") - checkNestedFieldEvolutionFallback( - s"SELECT id, s FROM $deepTable ORDER BY id", - "s.l.element.b (added), s.t (added)") + spark.sql( + s"INSERT INTO $deepTable VALUES (2, named_struct('l', array(named_struct('a', 2, 'b', 20)), 't', named_struct('x', 5)))") + checkIcebergNativeScan(s"SELECT id, s FROM $deepTable ORDER BY id") spark.sql(s"DROP TABLE $listTable") spark.sql(s"DROP TABLE $mapTable") @@ -2025,7 +2087,7 @@ class CometIcebergNativeSuite } } - test("schema evolution - nested field dropped and re-added under the same name falls back") { + test("schema evolution - nested field dropped and re-added under the same name stays native") { assume(icebergAvailable, "Iceberg not available in classpath") withTempIcebergDir { warehouseDir => @@ -2037,9 +2099,9 @@ class CometIcebergNativeSuite 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. + // The re-added s.b has a new field id, so the old file's s.b is not it and reads as NULL. + // 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) USING iceberg") spark.sql(s"INSERT INTO $table VALUES (1, named_struct('a', 1, 'b', 2))") @@ -2052,14 +2114,14 @@ class CometIcebergNativeSuite 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)") + checkIcebergNativeScan(s"SELECT id, s FROM $table ORDER BY id") spark.sql(s"DROP TABLE $table") } } } - test("schema evolution - nested field renamed falls back") { + test("schema evolution - nested field renamed stays native") { assume(icebergAvailable, "Iceberg not available in classpath") withTempIcebergDir { warehouseDir => @@ -2080,8 +2142,7 @@ class CometIcebergNativeSuite moved STRUCT ) 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. + // The nulls make Iceberg write the nested fields as optional, as the table declares them. spark.sql(s""" INSERT INTO $table VALUES (1, named_struct('a', 1, 'b', 'x'), array(named_struct('a', 1, 'b', 2)), @@ -2100,15 +2161,80 @@ class CometIcebergNativeSuite named_struct('b', 4, 'c', 3)) """) + checkIcebergNativeScan(s"SELECT id, s FROM $table ORDER BY id") + checkIcebergNativeScan(s"SELECT id, items FROM $table ORDER BY id") + checkIcebergNativeScan(s"SELECT id, moved FROM $table ORDER BY id") + + spark.sql(s"DROP TABLE $table") + } + } + } + + // Spark reads a struct whose current fields all postdate a data file as NULL at the top level and + // as a struct of NULLs inside another struct. The native read gives a struct of NULLs or NULL + // depending on what else the scan reads, so these reads fall back. + test("schema evolution - struct whose fields were all added 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_disjoint" + spark.sql(s""" + CREATE TABLE $table ( + id INT, + s STRUCT, + items ARRAY>, + outer_s STRUCT>, + partial STRUCT + ) USING iceberg + """) + spark.sql(s""" + INSERT INTO $table VALUES + (1, named_struct('a', 1), array(named_struct('a', 1)), + named_struct('x', 1, 'inner_s', named_struct('a', 1)), named_struct('a', 1, 'b', 2)), + (2, null, null, null, null) + """) + spark.sql(s"ALTER TABLE $table ADD COLUMN s.b INT") + spark.sql(s"ALTER TABLE $table DROP COLUMN s.a") + spark.sql(s"ALTER TABLE $table ADD COLUMN items.element.b INT") + spark.sql(s"ALTER TABLE $table DROP COLUMN items.element.a") + spark.sql(s"ALTER TABLE $table ADD COLUMN outer_s.inner_s.b INT") + spark.sql(s"ALTER TABLE $table DROP COLUMN outer_s.inner_s.a") + // partial keeps b, so a file that predates the change still has one of its fields. + spark.sql(s"ALTER TABLE $table ADD COLUMN partial.c INT") + spark.sql(s"ALTER TABLE $table DROP COLUMN partial.a") + spark.sql(s""" + INSERT INTO $table VALUES + (3, named_struct('b', 3), array(named_struct('b', 3)), + named_struct('x', 3, 'inner_s', named_struct('b', 3)), named_struct('b', 3, 'c', 4)) + """) + + checkNestedFieldEvolutionFallback(s"SELECT s FROM $table ORDER BY id", "s") + checkNestedFieldEvolutionFallback(s"SELECT id, s FROM $table ORDER BY id", "s") + checkNestedFieldEvolutionFallback(s"SELECT id, s.b FROM $table ORDER BY id", "s") + checkNestedFieldEvolutionFallback( + s"SELECT id FROM $table WHERE s IS NULL ORDER BY id", + "s") checkNestedFieldEvolutionFallback( - s"SELECT id, s FROM $table ORDER BY id", - "s.z (renamed from a)") + s"SELECT items FROM $table ORDER BY id", + "items.element") checkNestedFieldEvolutionFallback( - s"SELECT id, items FROM $table ORDER BY id", - "items.element.z (renamed from a)") + s"SELECT outer_s FROM $table ORDER BY id", + "outer_s.inner_s") checkNestedFieldEvolutionFallback( - s"SELECT id, moved FROM $table ORDER BY id", - "moved.c (renamed from a)") + s"SELECT id, outer_s, items, s FROM $table ORDER BY id", + "items.element, outer_s.inner_s, s") + + // A read that does not project those columns, or whose struct kept a field, stays native. + checkIcebergNativeScan(s"SELECT id FROM $table ORDER BY id") + checkIcebergNativeScan(s"SELECT id, partial FROM $table ORDER BY id") spark.sql(s"DROP TABLE $table") } @@ -2908,13 +3034,19 @@ class CometIcebergNativeSuite } rows } - // `s.b` is the struct's last field, so a positional cast would read `s.a` instead. + // Without field ids or a name mapping, iceberg-rust matches a struct's fields to the + // table's by position, and only when their counts are equal. A read of the whole struct + // gets the file's values, as it did before. A pruned read of some of its fields fails + // rather than reading a different field, since `s.b` is the struct's last field. Seq( s"SELECT id, s.b FROM $table ORDER BY id", s"SELECT id, items.x FROM $table ORDER BY id").foreach { query => val full = nativeRows(query, pruneNestedFields = false) assert(full.length == 100 && full.forall(!_.isNullAt(1)), s"$query: $full") - assert(nativeRows(query, pruneNestedFields = true) == full, query) + val e = intercept[SparkException](nativeRows(query, pruneNestedFields = true)) + assert( + e.getMessage.contains("cannot reconcile struct fields by id"), + s"$query: ${e.getMessage}") } spark.sql(s"DROP TABLE $table") diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergTestBase.scala b/spark/src/test/scala/org/apache/comet/CometIcebergTestBase.scala index 179a53e0915..f85917bb142 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergTestBase.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergTestBase.scala @@ -124,13 +124,16 @@ trait CometIcebergTestBase { this: CometTestBase => /** * Adds an `int` column whose initial and write defaults are both `defaultValue`, through - * `UpdateSchema.addColumn(name, type, Literal)`. Requires Iceberg 1.10+ and a v3 table. Callers - * must `REFRESH TABLE` afterwards. + * `UpdateSchema.addColumn(name, type, Literal)`, or to the struct `parent` through + * `addColumn(parent, name, type, Literal)`. Requires Iceberg 1.10+ and a v3 table. Callers must + * `REFRESH TABLE` afterwards. */ protected def addIcebergIntColumnWithDefault( icebergTable: AnyRef, columnName: String, - defaultValue: Int): Unit = { + defaultValue: Int, + parent: Option[String] = None): Unit = { + val typeClass = IcebergReflection.loadClass("org.apache.iceberg.types.Type") val intType = IcebergReflection .loadClass("org.apache.iceberg.types.Types$IntegerType") .getMethod("get") @@ -139,18 +142,22 @@ trait CometIcebergTestBase { this: CometTestBase => .loadClass("org.apache.iceberg.expressions.Expressions") .getMethod("lit", classOf[Object]) .invoke(null, Integer.valueOf(defaultValue)) + val literalClass = IcebergReflection.loadClass("org.apache.iceberg.expressions.Literal") val update = IcebergReflection .loadClass("org.apache.iceberg.Table") .getMethod("updateSchema") .invoke(icebergTable) val updateSchemaClass = IcebergReflection.loadClass("org.apache.iceberg.UpdateSchema") - updateSchemaClass - .getMethod( - "addColumn", - classOf[String], - IcebergReflection.loadClass("org.apache.iceberg.types.Type"), - IcebergReflection.loadClass("org.apache.iceberg.expressions.Literal")) - .invoke(update, columnName, intType, literal) + parent match { + case Some(struct) => + updateSchemaClass + .getMethod("addColumn", classOf[String], classOf[String], typeClass, literalClass) + .invoke(update, struct, columnName, intType, literal) + case None => + updateSchemaClass + .getMethod("addColumn", classOf[String], typeClass, literalClass) + .invoke(update, columnName, intType, literal) + } updateSchemaClass.getMethod("commit").invoke(update) }