From 6e44a5857e46de9b0508ccb8d190d593e351304b Mon Sep 17 00:00:00 2001 From: Andy Grove Date: Mon, 7 Sep 2026 07:12:04 -0600 Subject: [PATCH] fix: read Iceberg tables partitioned by an unknown transform Iceberg Java surfaces a transform it does not recognize as an UnknownTransform whose toString is the original name (e.g. "zero") and whose result type is string, not unknown -- so serializePartitionData's unknown-type filter keeps the field and the spec reaches native with a transform name iceberg-rust cannot deserialize, alongside a real partition value. The task then has partition values and no spec, which FileScanTask's validation rejects. Map such a name onto "unknown", which iceberg-rust deserializes into Transform::Unknown -- its model of the same thing, with the same string result type Iceberg Java reports, so the partition type serialized alongside the spec still agrees with it. Safe because the transform name reaches nothing in the native read but the identity test that builds the partition constants map, and identity is matched exactly. --- native/core/src/execution/planner.rs | 141 ++++++++++++++++-- .../comet/iceberg/IcebergReflection.scala | 38 +++++ .../operator/CometIcebergNativeScan.scala | 9 +- .../comet/CometIcebergNativeSuite.scala | 120 ++++++++++++++- .../iceberg/IcebergReflectionSuite.scala | 51 +++++++ 5 files changed, 343 insertions(+), 16 deletions(-) diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs index 5cfe74aa412..8a0972d3977 100644 --- a/native/core/src/execution/planner.rs +++ b/native/core/src/execution/planner.rs @@ -4397,11 +4397,12 @@ fn parse_file_scan_tasks_from_common( )) })?; // Recover spec_id from the index-aligned spec JSON's "spec-id" field directly, - // rather than from partition_spec_cache: a spec that uses a transform iceberg-rust - // doesn't recognize (e.g. forward-compatibility tests) fails to deserialize into a - // PartitionSpec, but its "spec-id" is still present in the JSON and its partition - // type entry has no usable fields anyway. Falling back to idx keeps ordering - // deterministic if the field is somehow absent. + // rather than from partition_spec_cache: a spec that fails to deserialize into a + // PartitionSpec still has its "spec-id" in the JSON, so the merge below stays + // correctly ordered. The JVM serializer maps a transform name iceberg-rust cannot + // parse to "unknown" (see IcebergReflection.Transforms.forNative), so this is + // defensive rather than a path a supported table reaches. Falling back to idx keeps + // ordering deterministic if the field is somehow absent. let spec_id = proto_common .partition_spec_pool .get(idx) @@ -7191,11 +7192,11 @@ mod tests { #[test] fn test_unified_partition_type_tolerates_unparseable_spec() { - // Regression for TestForwardCompatibility.testSparkCanReadUnknownTransform: a spec that - // uses a transform iceberg-rust doesn't recognize fails to deserialize into a - // PartitionSpec, but its partition field is filtered out on the Scala side (unknown type), - // so its partition_type_pool entry is an empty struct. Recovering spec_id must not require - // the full spec to parse -- the merge must succeed (with no fields) rather than erroring. + // A spec whose transform iceberg-rust cannot deserialize still has to yield a spec_id for + // the unified-type merge, so recovering it must not require the full spec to parse. The JVM + // serializer no longer emits such a spec (see + // test_unknown_transform_spec_builds_task_with_partition_data), so this pins the defensive + // path: the merge succeeds rather than erroring. let schema_json = serde_json::to_string( &iceberg::spec::Schema::builder() .with_schema_id(0) @@ -7211,7 +7212,8 @@ mod tests { .expect("serialize schema"); // A spec JSON whose transform iceberg-rust cannot deserialize, paired with an empty type - // entry (as Scala produces once the unknown-type field is filtered). + // entry. No partition data: `FileScanTask` validation rejects partition values without a + // spec, and the unparseable spec deserializes to None. let unparseable_spec_json = r#"{"spec-id":7,"fields":[{"source-id":1,"field-id":1000,"name":"x","transform":"totally_unknown[9]"}]}"#; let empty_type_json = r#"{"type":"struct","fields":[]}"#; @@ -7245,4 +7247,121 @@ mod tests { "unified partition type should have no fields for an all-unknown-transform spec" ); } + + /// Regression for TestForwardCompatibility.testSparkCanReadUnknownTransform. Iceberg Java + /// reports a transform it doesn't know (written by a newer Iceberg) under its original name, + /// e.g. `zero`, and the field's partition type as `string` -- so the field is NOT dropped from + /// the spec, and the task carries a real partition value for it. If that name reached + /// iceberg-rust verbatim the spec would fail to deserialize, leaving partition values with no + /// spec, which `FileScanTask::build()` rejects with "Non-empty FileScanTask partition requires + /// a partition spec". `IcebergReflection.Transforms.forNative` maps it to `unknown`, which + /// iceberg-rust parses; this pins that the resulting task builds and keeps its spec. + #[test] + fn test_unknown_transform_spec_builds_task_with_partition_data() { + let schema_json = serde_json::to_string( + &iceberg::spec::Schema::builder() + .with_schema_id(0) + .with_fields(vec![ + iceberg::spec::NestedField::optional( + 1, + "id", + iceberg::spec::Type::Primitive(iceberg::spec::PrimitiveType::Long), + ) + .into(), + iceberg::spec::NestedField::optional( + 2, + "data", + iceberg::spec::Type::Primitive(iceberg::spec::PrimitiveType::String), + ) + .into(), + ]) + .build() + .expect("schema"), + ) + .expect("serialize schema"); + + // What the JVM serializer emits for TestForwardCompatibility's UNKNOWN_SPEC: the `zero` + // transform normalized to `unknown`, and the partition type Iceberg Java resolved for it + // (string, from UnknownTransform.getResultType). + let spec_json = r#"{"spec-id":0,"fields":[{"source-id":1,"field-id":1000,"name":"id_zero","transform":"unknown"}]}"#; + let type_json = r#"{"type":"struct","fields":[{"id":1000,"name":"id_zero","required":false,"type":"string"}]}"#; + + let proto_common = spark_operator::IcebergScanCommon { + schema_pool: vec![schema_json], + partition_type_pool: vec![type_json.to_string()], + partition_spec_pool: vec![spec_json.to_string()], + partition_data_pool: vec![spark_operator::PartitionData { + values: vec![spark_operator::PartitionValue { + field_id: 1000, + literal: Some(spark_operator::IcebergLiteral { + value: Some(spark_operator::iceberg_literal::Value::StringVal( + "0".to_string(), + )), + ..Default::default() + }), + }], + }], + project_field_ids_pool: vec![spark_operator::ProjectFieldIdList { + field_ids: vec![1, 2, iceberg::metadata_columns::RESERVED_FIELD_ID_PARTITION], + }], + ..Default::default() + }; + + let proto_task = spark_operator::IcebergFileScanTask { + data_file_path: "file:///tmp/data.parquet".to_string(), + file_size_in_bytes: 100, + schema_idx: 0, + partition_spec_idx: Some(0), + partition_data_idx: Some(0), + project_field_ids_idx: 0, + ..Default::default() + }; + + let tasks = + parse_file_scan_tasks_from_common(&proto_common, std::slice::from_ref(&proto_task)) + .expect("an unknown-transform spec must still build a scan task"); + assert_eq!(tasks.len(), 1); + + let spec = tasks[0] + .partition_spec() + .expect("partition spec must survive an unknown transform"); + assert_eq!(spec.spec_id(), 0); + assert_eq!(spec.fields().len(), 1); + assert_eq!( + spec.fields()[0].transform, + iceberg::spec::Transform::Unknown + ); + assert_eq!( + tasks[0].partition().map(|p| p.fields().len()), + Some(1), + "the partition value must be carried alongside the spec" + ); + + // The `_partition` column reads its field types from the JVM-supplied partition type, which + // agrees with Transform::Unknown's own result type (string). + let unified = tasks[0] + .unified_partition_type() + .expect("unified_partition_type must be set when _partition is projected"); + assert_eq!(unified.fields().len(), 1); + assert_eq!(unified.fields()[0].id, 1000); + assert_eq!( + *unified.fields()[0].field_type, + iceberg::spec::Type::Primitive(iceberg::spec::PrimitiveType::String) + ); + + // Pin the failure the normalization avoids: the raw Iceberg Java transform name is the same + // input in every other respect and does not survive `FileScanTask` validation. + let raw_spec_json = spec_json.replace(r#""transform":"unknown""#, r#""transform":"zero""#); + assert_ne!(raw_spec_json, spec_json, "the replace must have matched"); + let raw_common = spark_operator::IcebergScanCommon { + partition_spec_pool: vec![raw_spec_json], + ..proto_common + }; + let err = parse_file_scan_tasks_from_common(&raw_common, &[proto_task]) + .expect_err("an unnormalized transform name must not build a task"); + assert!( + err.to_string().contains("Non-empty FileScanTask partition"), + "unexpected error: {err}" + ); + } } 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 fb705ce14b4..e1ac9993edd 100644 --- a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala +++ b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala @@ -136,6 +136,44 @@ object IcebergReflection extends Logging { */ object Transforms { val IDENTITY = "identity" + + /** + * iceberg-rust's placeholder for a transform it does not recognize. Conservative by + * construction: it contributes no partition constants and no pruning. + */ + val UNKNOWN = "unknown" + + /** Transforms whose `toString` iceberg-rust's `Transform::from_str` matches exactly. */ + private val ExactNativeTransforms = + Set(IDENTITY, UNKNOWN, "void", "year", "month", "day", "hour") + + /** `bucket[N]` / `truncate[W]`, the two parameterized spellings that parser also accepts. */ + private val ParameterizedNativeTransform = """^(?:bucket|truncate)\[\d+\]$""".r + + /** + * Maps an Iceberg Java transform name onto one iceberg-rust can deserialize. + * + * Iceberg Java parses a transform it doesn't know (one written by a newer Iceberg) into an + * `UnknownTransform` whose `toString` is the original name, e.g. `zero`. Serializing that + * name verbatim makes `PartitionSpec` deserialization fail native-side, which leaves the scan + * task holding partition values with no spec -- rejected by `FileScanTask`'s validation, so + * the whole scan dies instead of reading a table Iceberg considers forward-compatible. + * `Transform::Unknown` is iceberg-rust's model of the same thing, and its result type + * (string) is what Iceberg Java's `UnknownTransform.getResultType` reports, so the partition + * type serialized alongside the spec still agrees with it. + * + * Rewriting is only safe because the transform name reaches nothing in the native read but + * the identity test that builds the partition constants map (`_spec_id` uses the spec id, + * `_partition` matches partition values by field id). `IDENTITY` is matched exactly and so is + * never rewritten; every other transform yields no constants either way. + */ + def forNative(transform: String): String = + if (ExactNativeTransforms.contains(transform) || + ParameterizedNativeTransform.pattern.matcher(transform).matches()) { + transform + } else { + UNKNOWN + } } /** diff --git a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala index d3e1d2ced67..44b7497c3ef 100644 --- a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala +++ b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala @@ -438,10 +438,11 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit .getMethod(specField.getClass, "name") .invoke(specField) .asInstanceOf[String] - val transform = IcebergReflection - .getMethod(specField.getClass, "transform") - .invoke(specField) - .toString + val transform = IcebergReflection.Transforms.forNative( + IcebergReflection + .getMethod(specField.getClass, "transform") + .invoke(specField) + .toString) ("source-id" -> sourceId) ~ ("field-id" -> fieldId) ~ ("name" -> name) ~ diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala index 2e6dd310334..6a04d892bfd 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala @@ -41,7 +41,7 @@ import org.apache.spark.sql.execution.exchange.{ReusedExchangeExec, ShuffleExcha import org.apache.spark.sql.internal.SQLConf import org.apache.spark.sql.types.{StringType, TimestampType} -import org.apache.comet.CometSparkSessionExtensions.{isSpark35Plus, isSpark40Plus, isSpark42Plus} +import org.apache.comet.CometSparkSessionExtensions.{isSpark35Plus, isSpark40Plus, isSpark41Plus, isSpark42Plus} import org.apache.comet.iceberg.{IcebergReflection, RESTCatalogHelper} import org.apache.comet.serde.OperatorOuterClass import org.apache.comet.testing.{FuzzDataGenerator, SchemaGenOptions} @@ -5765,4 +5765,122 @@ class CometIcebergNativeSuite } } } + + test("forward compatibility - read a table whose spec uses an unknown transform") { + // Iceberg's forward-compatibility contract: a reader must be able to read a table partitioned + // by a transform it does not know (one a newer writer produced). Iceberg Java parses such a + // transform into an UnknownTransform whose toString is the original name, e.g. "zero", and + // resolves its partition type as string -- so the field is NOT dropped from the spec and the + // scan task carries a real partition value for it. Serializing that name verbatim makes + // PartitionSpec deserialization fail in iceberg-rust, leaving the task holding partition values + // with no spec, which FileScanTask validation rejects ("Non-empty FileScanTask partition + // requires a partition spec") and the whole scan dies. + // Upstream coverage is TestForwardCompatibility.testSparkCanReadUnknownTransform, which builds + // the table through Iceberg's low-level manifest writers; here the same shape is reached by + // writing an identity-partitioned table and then rewriting its spec's transform, which keeps + // the test on APIs that are stable across the Iceberg versions Comet builds against. + assume(icebergAvailable, "Iceberg not available in classpath") + // Spark 4.1 validates a V2 relation's metadata columns on every read, which forces + // SparkTable.metadataColumns() -> Partitioning.partitionType() and rejects an unknown transform + // in the analyzer, with or without Comet. Upstream disabled its own copy of this test on 4.1 for + // the same reason (SPARK-55626), so there is no read left to accelerate there. + assume( + !isSpark41Plus, + "SPARK-55626: Spark 4.1+ cannot read a table with an unknown transform") + + withTempIcebergDir { warehouseDir => + withSQLConf( + "spark.sql.catalog.fwd_cat" -> "org.apache.iceberg.spark.SparkCatalog", + "spark.sql.catalog.fwd_cat.type" -> "hadoop", + "spark.sql.catalog.fwd_cat.warehouse" -> warehouseDir.getAbsolutePath, + // The rewrite below edits metadata behind the catalog's back, so don't let it serve the + // pre-rewrite TableMetadata from cache. + "spark.sql.catalog.fwd_cat.cache-enabled" -> "false", + CometConf.COMET_ENABLED.key -> "true", + CometConf.COMET_EXEC_ENABLED.key -> "true", + CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") { + + val table = "fwd_cat.db.unknown_transform" + try { + spark.sql(s""" + CREATE TABLE $table (id BIGINT, data STRING) + USING iceberg PARTITIONED BY (id) + """) + spark.sql(s""" + INSERT INTO $table + VALUES (1, 'a'), (2, 'b'), (2, 'c'), (3, NULL) + """) + + import org.apache.iceberg.catalog.TableIdentifier + import org.apache.iceberg.spark.SparkCatalog + + val sparkCatalog = spark.sessionState.catalogManager + .catalog("fwd_cat") + .asInstanceOf[SparkCatalog] + val iceTable = sparkCatalog + .icebergCatalog() + .loadTable(TableIdentifier.of("db", "unknown_transform")) + + val tableLocationUri = iceTable.location() + val tableDir = + if (tableLocationUri.contains(":")) new File(new java.net.URI(tableLocationUri)) + else new File(tableLocationUri) + val metadataDir = new File(tableDir, "metadata") + val versionPattern = "^v(\\d+)\\.metadata\\.json$".r + val currentVersion = metadataDir + .listFiles() + .flatMap(f => versionPattern.findFirstMatchIn(f.getName).map(_.group(1).toInt)) + .max + val currentMetadataFile = new File(metadataDir, s"v$currentVersion.metadata.json") + val currentMetadataJson = + new String(java.nio.file.Files.readAllBytes(currentMetadataFile.toPath), UTF_8) + + val mapper = new com.fasterxml.jackson.databind.ObjectMapper() + val root = mapper + .readTree(currentMetadataJson) + .asInstanceOf[com.fasterxml.jackson.databind.node.ObjectNode] + // Retarget the one spec the data was written under at a transform no Iceberg release + // defines. The manifests keep the partition values the identity spec recorded, exactly as + // upstream's fake-spec manifest does. + val specFields = root + .get("partition-specs") + .asInstanceOf[com.fasterxml.jackson.databind.node.ArrayNode] + .elements() + .asScala + .flatMap(_.get("fields").elements().asScala) + .toSeq + assert(specFields.size == 1, s"expected one partition field, got $specFields") + specFields.foreach( + _.asInstanceOf[com.fasterxml.jackson.databind.node.ObjectNode] + .put("transform", "zero")) + + val newMetadataJson = mapper.writeValueAsString(root) + // Round-trip through Iceberg's own parser so a malformed edit fails here rather than + // producing an unloadable table, and so this test also pins that Iceberg still accepts an + // unknown transform at load time (the premise of the whole scenario). + val reparsed = org.apache.iceberg.TableMetadataParser.fromJson(newMetadataJson) + assert( + reparsed.spec().fields().get(0).transform().toString == "zero", + "Iceberg no longer preserves an unknown transform's name") + + val newVersion = currentVersion + 1 + java.nio.file.Files.write( + new File(metadataDir, s"v$newVersion.metadata.json").toPath, + newMetadataJson.getBytes(UTF_8)) + java.nio.file.Files.write( + new File(metadataDir, "version-hint.text").toPath, + newVersion.toString.getBytes(UTF_8)) + + // Read by path, with no projection on top, for the same reason upstream's test does: + // resolving a Project's metadataOutput forces SparkTable.metadataColumns(), whose + // _partition type comes from Partitioning.partitionType() -- which rejects an unknown + // transform outright, in Spark's analyzer, with or without Comet. A bare relation scan is + // therefore the whole of what any reader can do with such a table. + checkIcebergNativeScan(spark.read.format("iceberg").load(tableDir.getAbsolutePath)) + } finally { + spark.sql(s"DROP TABLE IF EXISTS $table PURGE") + } + } + } + } } diff --git a/spark/src/test/scala/org/apache/comet/iceberg/IcebergReflectionSuite.scala b/spark/src/test/scala/org/apache/comet/iceberg/IcebergReflectionSuite.scala index 1fcfb010b22..a4831411df2 100644 --- a/spark/src/test/scala/org/apache/comet/iceberg/IcebergReflectionSuite.scala +++ b/spark/src/test/scala/org/apache/comet/iceberg/IcebergReflectionSuite.scala @@ -28,6 +28,7 @@ import org.apache.iceberg.BaseMetastoreTableOperations import org.apache.iceberg.BaseTable import org.apache.iceberg.DataFiles import org.apache.iceberg.PartitionSpec +import org.apache.iceberg.PartitionSpecParser import org.apache.iceberg.Schema import org.apache.iceberg.TableMetadata import org.apache.iceberg.io.FileIO @@ -240,6 +241,56 @@ class IcebergReflectionSuite extends AnyFunSuite { IcebergReflection.executorReflectionUnresolved) } + /** Schema the transform tests below partition on, one column per transform source type. */ + private val transformSchema = new Schema( + Types.NestedField.optional(1, "id", Types.LongType.get()), + Types.NestedField.optional(2, "s", Types.StringType.get()), + Types.NestedField.optional(3, "ts", Types.TimestampType.withZone())) + + /** + * The single-field spec Iceberg parses out of `transform`, applied to source column `sourceId`. + */ + private def singleFieldSpec(transform: String, sourceId: Int): PartitionSpec = + PartitionSpecParser.fromJson( + transformSchema, + s"""{"spec-id":0,"fields":[{"name":"p","transform":"$transform",""" + + s""""source-id":$sourceId,"field-id":1000}]}""") + + test("forNative keeps every transform iceberg-rust can deserialize") { + // Spelled as Iceberg serializes them. The round-trip assertion matters as much as forNative's + // own answer: forNative matches on Transform.toString, so a version that renders a transform + // differently from its JSON spelling would silently start rewriting it to "unknown". + Seq( + ("identity", 1), + ("void", 1), + ("bucket[8]", 1), + ("truncate[4]", 2), + ("year", 3), + ("month", 3), + ("day", 3), + ("hour", 3)).foreach { case (transform, sourceId) => + val rendered = singleFieldSpec(transform, sourceId).fields().get(0).transform().toString + assert(rendered == transform, s"Iceberg renders $transform as $rendered") + assert(IcebergReflection.Transforms.forNative(rendered) == transform) + } + } + + test("forNative rewrites a transform Iceberg Java could not resolve") { + // TestForwardCompatibility's UNKNOWN_SPEC. Iceberg parses "zero" into an UnknownTransform whose + // toString is the original name; serialized verbatim it fails PartitionSpec deserialization in + // iceberg-rust, leaving the scan task holding partition values with no spec, which + // FileScanTask validation rejects ("Non-empty FileScanTask partition requires a partition + // spec") and the whole scan dies. + val spec = singleFieldSpec("zero", 1) + val rendered = spec.fields().get(0).transform().toString + assert(rendered == "zero") + assert(IcebergReflection.Transforms.forNative(rendered) == "unknown") + + // The partition type Comet serializes alongside the rewritten spec has to agree with what + // iceberg-rust derives for Transform::Unknown, which is string. + assert(spec.partitionType().fields().get(0).`type`().toString == "string") + } + /** Mimics a newer Iceberg ContentFile, which exposes location(). */ class LocationFile(loc: String) { def location(): String = loc