From d062dd1111d46ae1cfe3d75e968206cb870e0ea1 Mon Sep 17 00:00:00 2001 From: Parth Chandra Date: Mon, 15 Jun 2026 13:20:11 -0700 Subject: [PATCH 1/6] feat: [iceberg] support iceberg metadata columns _pos, _spec, _file, _partition --- dev/diffs/iceberg/1.11.0.diff | 14 + native/Cargo.lock | 4 +- native/Cargo.toml | 4 +- .../src/execution/operators/iceberg_scan.rs | 1 + native/core/src/execution/planner.rs | 61 ++ .../comet/iceberg/IcebergReflection.scala | 36 + .../apache/comet/rules/CometScanRule.scala | 110 ++- .../operator/CometIcebergNativeScan.scala | 81 +- .../iceberg/metadata_column_partition.sql | 850 ++++++++++++++++++ .../comet/CometIcebergNativeSuite.scala | 491 +++++++++- 10 files changed, 1597 insertions(+), 55 deletions(-) create mode 100644 spark/src/test/resources/sql-tests/iceberg/metadata_column_partition.sql diff --git a/dev/diffs/iceberg/1.11.0.diff b/dev/diffs/iceberg/1.11.0.diff index ca56ab9b93..a8dc4d3150 100644 --- a/dev/diffs/iceberg/1.11.0.diff +++ b/dev/diffs/iceberg/1.11.0.diff @@ -57,6 +57,20 @@ index f766fbb79a..c5e31185a9 100644 .config(TestBase.DISABLE_UI) .enableHiveSupport() .getOrCreate(); +diff --git a/spark/v4.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/SparkRowLevelOperationsTestBase.java b/spark/v4.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/SparkRowLevelOperationsTestBase.java +index b5d6415763..0763a723d3 100644 +--- a/spark/v4.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/SparkRowLevelOperationsTestBase.java ++++ b/spark/v4.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/SparkRowLevelOperationsTestBase.java +@@ -448,7 +448,8 @@ public abstract class SparkRowLevelOperationsTestBase extends ExtensionsTestBase + + protected void assertAllBatchScansVectorized(SparkPlan plan) { + List batchScans = SparkPlanUtil.collectBatchScans(plan); +- assertThat(batchScans).hasSizeGreaterThan(0).allMatch(SparkPlan::supportsColumnar); ++ // When Comet is enabled, its native scan replaces BatchScanExec nodes entirely ++ assertThat(batchScans).allMatch(SparkPlan::supportsColumnar); + } + + protected void createTableWithDeleteGranularity( diff --git a/spark/v4.1/spark/src/jmh/java/org/apache/iceberg/spark/action/DeleteOrphanFilesBenchmark.java b/spark/v4.1/spark/src/jmh/java/org/apache/iceberg/spark/action/DeleteOrphanFilesBenchmark.java index 3fd84553f0..350b4a7561 100644 --- a/spark/v4.1/spark/src/jmh/java/org/apache/iceberg/spark/action/DeleteOrphanFilesBenchmark.java diff --git a/native/Cargo.lock b/native/Cargo.lock index d7b01acfcc..32beab0090 100644 --- a/native/Cargo.lock +++ b/native/Cargo.lock @@ -3496,7 +3496,7 @@ dependencies = [ [[package]] name = "iceberg" version = "0.10.0" -source = "git+https://github.com/apache/iceberg-rust?rev=4532da8#4532da80f3930cdbda1cf2effe05b50165da9875" +source = "git+https://github.com/apache/iceberg-rust?rev=3d84c81353b1b23b6e4ae8eea8f8a021cc6927a7#3d84c81353b1b23b6e4ae8eea8f8a021cc6927a7" dependencies = [ "aes-gcm", "anyhow", @@ -3552,7 +3552,7 @@ dependencies = [ [[package]] name = "iceberg-storage-opendal" version = "0.10.0" -source = "git+https://github.com/apache/iceberg-rust?rev=4532da8#4532da80f3930cdbda1cf2effe05b50165da9875" +source = "git+https://github.com/apache/iceberg-rust?rev=3d84c81353b1b23b6e4ae8eea8f8a021cc6927a7#3d84c81353b1b23b6e4ae8eea8f8a021cc6927a7" dependencies = [ "anyhow", "async-trait", diff --git a/native/Cargo.toml b/native/Cargo.toml index fd7d7d3a66..e41b3a1265 100644 --- a/native/Cargo.toml +++ b/native/Cargo.toml @@ -58,8 +58,8 @@ object_store = { version = "0.13.2", features = ["gcp", "azure", "aws", "http"] url = "2.2" aws-config = "1.8.18" aws-credential-types = "1.2.13" -iceberg = { git = "https://github.com/apache/iceberg-rust", rev = "4532da8" } -iceberg-storage-opendal = { git = "https://github.com/apache/iceberg-rust", rev = "4532da8", features = ["opendal-memory", "opendal-fs", "opendal-s3", "opendal-gcs", "opendal-oss", "opendal-azdls"] } +iceberg = { git = "https://github.com/apache/iceberg-rust", rev = "3d84c81353b1b23b6e4ae8eea8f8a021cc6927a7" } +iceberg-storage-opendal = { git = "https://github.com/apache/iceberg-rust", rev = "3d84c81353b1b23b6e4ae8eea8f8a021cc6927a7", 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/operators/iceberg_scan.rs b/native/core/src/execution/operators/iceberg_scan.rs index d77b208964..e727294fd9 100644 --- a/native/core/src/execution/operators/iceberg_scan.rs +++ b/native/core/src/execution/operators/iceberg_scan.rs @@ -660,6 +660,7 @@ mod tests { partition: None, partition_spec: None, name_mapping: None, + unified_partition_type: None, case_sensitive: false, key_metadata: None, } diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs index 39e7ccc2e5..116c11770f 100644 --- a/native/core/src/execution/planner.rs +++ b/native/core/src/execution/planner.rs @@ -3835,6 +3835,41 @@ fn parse_file_scan_tasks_from_common( }) .collect::, _>>()?; + // Compute unified partition type from the partition_type_pool. + // Each entry is a StructType JSON with the resolved field types for one partition spec. + // Merge all specs into a single unified type (dedup by field_id). + // + // NOTE: The type information here comes from Iceberg Java's PartitionSpec.partitionType() + // (via Scala reflection). This is the same type computation as iceberg-rust's + // Transform::result_type() -- both implement the Iceberg spec's transform result rules. + // When iceberg-rust's own scan planning is used (not Comet's proto path), it computes + // this via compute_unified_partition_type(specs, schema) instead. + let unified_partition_type = { + let mut seen_field_ids = std::collections::HashSet::new(); + let mut struct_fields: Vec = Vec::new(); + + for type_json in &proto_common.partition_type_pool { + match serde_json::from_str::(type_json) { + Ok(struct_type) => { + for field in struct_type.fields() { + if seen_field_ids.insert(field.id) { + struct_fields.push(Arc::clone(field)); + } + } + } + Err(e) => { + return Err(ExecutionError::GeneralError(format!( + "Failed to deserialize partition type JSON from pool: {e}" + ))); + } + } + } + + iceberg::spec::StructType::new(struct_fields) + }; + + let unified_partition_type_arc = Arc::new(unified_partition_type); + let results: Result, _> = proto_tasks .iter() .map(|proto_task| { @@ -3939,6 +3974,14 @@ fn parse_file_scan_tasks_from_common( .field_ids .clone(); + let unified_partition_type_for_task = if project_field_ids + .contains(&iceberg::metadata_columns::RESERVED_FIELD_ID_PARTITION) + { + Some(Arc::clone(&unified_partition_type_arc)) + } else { + None + }; + Ok(iceberg::scan::FileScanTask { file_size_in_bytes: proto_task.file_size_in_bytes, data_file_path: proto_task.data_file_path.clone(), @@ -3953,6 +3996,7 @@ fn parse_file_scan_tasks_from_common( partition, partition_spec, name_mapping, + unified_partition_type: unified_partition_type_for_task, case_sensitive: false, // Plaintext StandardKeyMetadata forwarded verbatim from the JVM; decoded by // iceberg-rust with no KMS unwrap. None for unencrypted data files. @@ -5455,4 +5499,21 @@ mod tests { } }); } + + #[test] + fn test_metadata_field_id_constants_match_iceberg_rust() { + // These constants are duplicated in Scala (CometIcebergNativeScan.MetadataFieldIds) + // as Int.MaxValue - 1 and Int.MaxValue - 5. This test ensures the Rust constants + // haven't drifted, which would cause a silent mismatch with the Scala side. + assert_eq!( + iceberg::metadata_columns::RESERVED_FIELD_ID_FILE, + i32::MAX - 1, + "RESERVED_FIELD_ID_FILE must be i32::MAX - 1 to match Scala MetadataFieldIds" + ); + assert_eq!( + iceberg::metadata_columns::RESERVED_FIELD_ID_PARTITION, + i32::MAX - 5, + "RESERVED_FIELD_ID_PARTITION must be i32::MAX - 5 to match Scala MetadataFieldIds" + ); + } } 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 72e885e834..c06b1b4e47 100644 --- a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala +++ b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala @@ -52,6 +52,8 @@ object IcebergReflection extends Logging { val SPARK_BATCH_QUERY_SCAN = "org.apache.iceberg.spark.source.SparkBatchQueryScan" val SPARK_STAGED_SCAN = "org.apache.iceberg.spark.source.SparkStagedScan" val SPARK_SCHEMA_UTIL = "org.apache.iceberg.spark.SparkSchemaUtil" + val TABLE = "org.apache.iceberg.Table" + val PARTITIONING = "org.apache.iceberg.Partitioning" } /** @@ -461,6 +463,40 @@ object IcebergReflection extends Logging { } } + /** + * Validates that the table's unified partition type can be computed -- the merge of every + * historical partition spec, which is what the `_partition` metadata column projects. + * + * Iceberg Java's `Partitioning.partitionType(table)` runs the same cross-spec compatibility + * check that iceberg-rust does natively: a V1 table does not guarantee partition field ids are + * unique across specs, so two specs can bind the same id to incompatible source/transform + * pairs, which cannot be merged into one struct field. iceberg-rust returns a DataInvalid error + * in that case, but only at scan time -- too late for Comet to fall back. Calling the Java + * check here, at plan time, lets `CometScanRule` fall back to Spark instead of failing inside + * the native reader. + * + * Returns None when the unified type is computable, or Some(reason) when it is not -- either + * the specs conflict or the reflection call itself failed. Both mean Comet cannot safely serve + * `_partition`, so both map to a fallback. + */ + def validateUnifiedPartitionType(table: Any): Option[String] = { + try { + val tableClass = loadClass(ClassNames.TABLE) + val partitioningClass = loadClass(ClassNames.PARTITIONING) + partitioningClass + .getMethod("partitionType", tableClass) + .invoke(null, table.asInstanceOf[AnyRef]) + None + } catch { + // A conflict surfaces as the ValidationException thrown by partitionType(), wrapped by + // reflection in InvocationTargetException; unwrap it for a meaningful reason. + case e: java.lang.reflect.InvocationTargetException => + Some(Option(e.getCause).getOrElse(e).getMessage) + case e: Exception => + Some(e.getMessage) + } + } + /** * Gets the table metadata from an Iceberg table. * 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 5e451b382a..b936a91c18 100644 --- a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala +++ b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala @@ -88,12 +88,6 @@ case class CometScanRule(session: SparkSession) case _ => false } - def metadataCols(plan: SparkPlan): Seq[String] = { - plan.expressions.collect { - case a: Attribute if a.isMetadataCol => a.name - } - } - def isIcebergMetadataTable(scanExec: BatchScanExec): Boolean = { // List of Iceberg metadata tables: // https://iceberg.apache.org/docs/latest/spark-queries/#inspecting-tables @@ -130,11 +124,6 @@ case class CometScanRule(session: SparkSession) case scan if !CometConf.COMET_NATIVE_SCAN_ENABLED.get(conf) => withFallbackReason(scan, "Comet Scan is not enabled") - case scan if metadataCols(scan).nonEmpty => - withFallbackReason( - scan, - s"Metadata column(s) ${metadataCols(scan).mkString(", ")} is not supported") - // data source V1 case scanExec: FileSourceScanExec => transformV1Scan(fullPlan, scanExec) @@ -154,6 +143,12 @@ case class CometScanRule(session: SparkSession) } private def transformV1Scan(plan: SparkPlan, scanExec: FileSourceScanExec): SparkPlan = { + val metadataColNames = metadataCols(scanExec) + if (metadataColNames.nonEmpty) { + return withFallbackReason( + scanExec, + s"Metadata column(s) ${metadataColNames.mkString(", ")} is not supported") + } // On Spark 3.4, injectQueryStageOptimizerRule is unavailable, so // CometPlanAdaptiveDynamicPruningFilters cannot run. Fall back this scan to Spark so that @@ -302,6 +297,11 @@ case class CometScanRule(session: SparkSession) scanExec.scan match { case scan: CSVScan if COMET_CSV_V2_NATIVE_ENABLED.get() => + if (scanExec.output.exists(_.isMetadataCol)) { + return withFallbackReason( + scanExec, + "Metadata columns are not supported for CSV V2 scans") + } val fallbackReasons = new ListBuffer[String]() val schemaSupported = CometBatchScanExec.isSchemaSupported(scan.readDataSchema, fallbackReasons) @@ -361,9 +361,30 @@ case class CometScanRule(session: SparkSession) return withFallbackReasons(scanExec, fallbackReasons.toSet) } + // Check for unsupported metadata columns in Iceberg scans + val unsupportedMetadataCols = scanExec.output.filter(_.isMetadataCol).filterNot { attr => + CometIcebergNativeScan.MetadataFieldIds.keySet.contains(attr.name) + } + if (unsupportedMetadataCols.nonEmpty) { + fallbackReasons += "Unsupported Iceberg metadata columns: " + + unsupportedMetadataCols.map(_.name).mkString(", ") + return withFallbackReasons(scanExec, fallbackReasons.toSet) + } + val typeChecker = CometScanTypeChecker() + // Filter out metadata columns from schema check -- their types are handled + // by iceberg-rust directly (e.g., _partition can be an empty struct for + // unpartitioned tables which the general type checker rejects). + val metadataColNames = scanExec.output.filter(_.isMetadataCol).map(_.name).toSet + val dataSchema = if (metadataColNames.nonEmpty) { + val filtered = + scanExec.scan.readSchema().filter(f => !metadataColNames.contains(f.name)) + new org.apache.spark.sql.types.StructType(filtered.toArray) + } else { + scanExec.scan.readSchema() + } val schemaSupported = - typeChecker.isSchemaSupported(scanExec.scan.readSchema(), fallbackReasons) + typeChecker.isSchemaSupported(dataSchema, fallbackReasons) if (!schemaSupported) { fallbackReasons += "Comet extension is not enabled for " + @@ -451,15 +472,17 @@ case class CometScanRule(session: SparkSession) } // Now perform all validation using the pre-extracted metadata - // Check if table uses a FileIO implementation compatible with iceberg-rust - + // Check if table uses a FileIO implementation compatible with iceberg-rust. + // Comet's native reader uses object_store (Rust) for I/O, bypassing Iceberg Java's + // FileIO entirely. Only allow known-compatible implementations whose underlying + // storage object_store can reach via standard URL schemes. val fileIOCompatible = IcebergReflection.getFileIO(metadata.table) match { - case Some(fileIO) - if fileIO.getClass.getName == "org.apache.iceberg.inmemory.InMemoryFileIO" => - fallbackReasons += "InMemoryFileIO is not supported by Comet's native reader" - false - case Some(_) => + case Some(fileIO) if CometScanRule.isCompatibleFileIO(fileIO.getClass.getName) => true + case Some(fileIO) => + fallbackReasons += s"FileIO ${fileIO.getClass.getName} is not supported by " + + "Comet's native reader (object_store bypasses Iceberg Java FileIO)" + false case None => fallbackReasons += "Could not check FileIO compatibility" false @@ -603,6 +626,25 @@ case class CometScanRule(session: SparkSession) false } + // The `_partition` metadata column projects the table's unified partition type -- the + // merge of all historical specs. iceberg-rust computes that merge at scan time and errors + // (DataInvalid) when specs bind one field id to incompatible source/transform pairs (only + // possible in V1 tables, which do not keep partition field ids unique across specs). That + // error would fail the native scan with no chance to fall back, so when `_partition` is + // projected we run Iceberg Java's equivalent check here and fall back if it cannot merge. + val unifiedPartitionTypeSupported = + if (scanExec.output.exists(a => a.isMetadataCol && a.name == "_partition")) { + IcebergReflection.validateUnifiedPartitionType(metadata.table) match { + case Some(reason) => + fallbackReasons += "Iceberg table has partition specs whose unified partition " + + s"type cannot be computed for the _partition metadata column: $reason" + false + case None => true + } + } else { + true + } + // Get filter expressions for complex predicates check val filterExpressionsOpt = IcebergReflection.getFilterExpressions(scanExec.scan) @@ -788,6 +830,7 @@ case class CometScanRule(session: SparkSession) if (schemaSupported && fileIOCompatible && formatVersionSupported && defaultValuesSupported && schemaTypesSupported && encryptionKeyLengthSupported && taskValidation.allParquet && allSupportedFilesystems && partitionTypesSupported && + unifiedPartitionTypeSupported && complexTypePredicatesSupported && transformFunctionsSupported && deleteFileTypesSupported && dppSubqueriesSupported) { CometBatchScanExec( @@ -824,6 +867,12 @@ case class CometScanRule(session: SparkSession) case _ => false } + private def metadataCols(plan: SparkPlan): Seq[String] = { + plan.expressions.collect { + case a: Attribute if a.isMetadataCol => a.name + } + } + private def isSchemaSupported(scanExec: FileSourceScanExec, r: HadoopFsRelation): Boolean = { val fallbackReasons = new ListBuffer[String]() val typeChecker = CometScanTypeChecker() @@ -882,6 +931,29 @@ case class CometScanTypeChecker() extends DataTypeSupport with CometTypeShim { object CometScanRule extends Logging { + // Iceberg FileIO implementations whose underlying storage object_store can reach. + // Custom/test FileIO classes (e.g. CustomFileIO in TestSparkExecutorCache) are not compatible + // because Comet's native reader bypasses Java FileIO entirely. + private val CompatibleFileIOClasses: Set[String] = Set( + "org.apache.iceberg.hadoop.HadoopFileIO", + "org.apache.iceberg.aws.s3.S3FileIO", + "org.apache.iceberg.gcp.gcs.GCSFileIO", + "org.apache.iceberg.io.ResolvingFileIO", + "org.apache.iceberg.spark.SparkFileIO", + "org.apache.iceberg.azure.adlsv2.ADLSFileIO", + "org.apache.iceberg.CachingFileIO") + + // Prefix of the EncryptingFileIO family. An encrypted table's io() is not the bare + // EncryptingFileIO but a nested variant chosen from the wrapped delegate's capabilities + // (e.g. EncryptingFileIO$WithSupportsPrefixOperations when the delegate is HadoopFileIO), so + // an exact class-name match misses it. Comet forwards each file's key_metadata to iceberg-rust + // and reads the ciphertext via object_store, so any EncryptingFileIO variant is compatible. + private val EncryptingFileIOPrefix = "org.apache.iceberg.encryption.EncryptingFileIO" + + /** True if `className` is a FileIO whose backing storage Comet's native reader can reach. */ + def isCompatibleFileIO(className: String): Boolean = + CompatibleFileIOClasses.contains(className) || className.startsWith(EncryptingFileIOPrefix) + // Per-scheme memo of `NativeBase.isObjectStoreSchemeSupported`. The answer depends only on the // URL scheme, so we cache by scheme and never re-cross the JNI boundary for a repeated scheme. private val schemeSupportCache = 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 c2b792c36c..42ef14e9fa 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 @@ -85,6 +85,20 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit Constants.Operations.GT -> OperatorOuterClass.IcebergPredicateOperator.GreaterThan, Constants.Operations.GT_EQ -> OperatorOuterClass.IcebergPredicateOperator.GreaterThanOrEq) + // Iceberg reserved field IDs for metadata columns. + // These must match iceberg-rust's constants in crates/iceberg/src/metadata_columns.rs: + // RESERVED_FIELD_ID_FILE = i32::MAX - 1 (2147483646) + // RESERVED_FIELD_ID_POS = i32::MAX - 2 (2147483645) + // RESERVED_FIELD_ID_SPEC_ID = i32::MAX - 4 (2147483643) + // RESERVED_FIELD_ID_PARTITION = i32::MAX - 5 (2147483642) + // Scala's Int.MaxValue == 2^31 - 1 == Rust's i32::MAX. + val MetadataFieldIds: Map[String, Int] = + Map( + "_file" -> (Int.MaxValue - 1), + "_pos" -> (Int.MaxValue - 2), + "_spec_id" -> (Int.MaxValue - 4), + "_partition" -> (Int.MaxValue - 5)) + /** * Wraps an Iceberg partition value (a typed primitive) in a PartitionValue. The value encoding * is shared with predicate literals via [[icebergLiteralToProto]]. @@ -487,25 +501,48 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit } }.toSeq - // Only serialize partition data if we have non-unknown fields - if (partitionValues.nonEmpty) { - val partitionDataProto = OperatorOuterClass.PartitionData - .newBuilder() - .addAllValues(partitionValues.asJava) - .build() - - // Deduplicate by protobuf bytes (use Base64 string as key) - val partitionDataBytes = partitionDataProto.toByteArray - val partitionDataKey = Base64.getEncoder.encodeToString(partitionDataBytes) - - val partitionDataIdx = partitionDataToPoolIndex.getOrElseUpdate( - partitionDataKey, { - val idx = partitionDataToPoolIndex.size - commonBuilder.addPartitionDataPool(partitionDataProto) + // Native requires a task to carry both a partition spec and partition data, or neither: + // iceberg-rust errors when the unified partition type has fields but a task is missing + // its spec/data. A file written while a partition field was dropped (partition + // evolution) has no partition values, so partitionValues is empty here. + // + // For that value-less case we must NOT send the file's real spec: it may retain a void + // field whose id collides with a unified field, which would make iceberg-rust index the + // empty partition data out of range. Instead send an empty-fields spec that keeps the + // real spec id (so _spec_id stays correct) plus empty data, so native fills every + // unified _partition field with null -- matching Spark and the pre-tightening behaviour. + if (partitionValues.isEmpty) { + val specId = spec.getClass.getMethod("specId").invoke(spec).asInstanceOf[Int] + val emptySpecJson = s"""{"spec-id":$specId,"fields":[]}""" + val emptySpecIdx = partitionSpecToPoolIndex.getOrElseUpdate( + emptySpecJson, { + val idx = partitionSpecToPoolIndex.size + commonBuilder.addPartitionSpecPool(emptySpecJson) idx }) - taskBuilder.setPartitionDataIdx(partitionDataIdx) + // Override the real spec registered above. + taskBuilder.setPartitionSpecIdx(emptySpecIdx) } + + // Always send partition data (empty when there are no values) so native never sees a + // spec without data. Native uses it to build the identity-transform constants_map and, + // together with the spec above, the _partition column. + val partitionDataProto = OperatorOuterClass.PartitionData + .newBuilder() + .addAllValues(partitionValues.asJava) + .build() + + // Deduplicate by protobuf bytes (use Base64 string as key) + val partitionDataBytes = partitionDataProto.toByteArray + val partitionDataKey = Base64.getEncoder.encodeToString(partitionDataBytes) + + val partitionDataIdx = partitionDataToPoolIndex.getOrElseUpdate( + partitionDataKey, { + val idx = partitionDataToPoolIndex.size + commonBuilder.addPartitionDataPool(partitionDataProto) + idx + }) + taskBuilder.setPartitionDataIdx(partitionDataIdx) } } } catch { @@ -1026,14 +1063,16 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit val nameToFieldId = IcebergReflection.buildFieldIdMapping(schema) - val projectFieldIds = output.flatMap { attr => + val projectFieldIds = output.map { attr => nameToFieldId .get(attr.name) .orElse(metadata.globalFieldIdMapping.get(attr.name)) - .orElse { - logWarning(s"Column '${attr.name}' not found in task or scan schema, " + - "skipping projection") - None + .orElse(CometIcebergNativeScan.MetadataFieldIds.get(attr.name)) + .getOrElse { + throw new IllegalStateException( + s"Column '${attr.name}' not found in task schema, global schema, " + + "or metadata field IDs. This indicates a bug in CometScanRule " + + "validation -- all output columns should be resolvable.") } } diff --git a/spark/src/test/resources/sql-tests/iceberg/metadata_column_partition.sql b/spark/src/test/resources/sql-tests/iceberg/metadata_column_partition.sql new file mode 100644 index 0000000000..7f226a036e --- /dev/null +++ b/spark/src/test/resources/sql-tests/iceberg/metadata_column_partition.sql @@ -0,0 +1,850 @@ +-- Licensed to the Apache Software Foundation (ASF) under one +-- or more contributor license agreements. See the NOTICE file +-- distributed with this work for additional information +-- regarding copyright ownership. The ASF licenses this file +-- to you under the Apache License, Version 2.0 (the +-- "License"); you may not use this file except in compliance +-- with the License. You may obtain a copy of the License at +-- +-- http://www.apache.org/licenses/LICENSE-2.0 +-- +-- Unless required by applicable law or agreed to in writing, +-- software distributed under the License is distributed on an +-- "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +-- KIND, either express or implied. See the License for the +-- specific language governing permissions and limitations +-- under the License. + +-- Native Iceberg scan is unsupported on Spark 4.2: no Iceberg spark-runtime is published for +-- 4.2 yet, so the build reuses the 4.0 runtime, which is binary-incompatible with 4.2 (e.g. +-- connector.catalog.View became a class). Mirrors the assume(!isSpark42Plus) guard in +-- CometIcebergNativeSuite. See https://github.com/apache/datafusion-comet/issues/4969. +-- MaxSparkVersion: 4.1 + +-- Config: spark.sql.catalog.test_cat=org.apache.iceberg.spark.SparkCatalog +-- Config: spark.sql.catalog.test_cat.type=hadoop +-- Config: spark.sql.catalog.test_cat.warehouse=/tmp/comet-iceberg-sql-test +-- Config: spark.comet.enabled=true +-- Config: spark.comet.exec.enabled=true +-- Config: spark.comet.iceberg.native.enabled=true + +-- All `query` assertions below implicitly validate that Comet's partition type +-- computation (from partition_type_pool JSON) matches Spark's (from Iceberg Java's +-- PartitionSpec.partitionType()). If the two diverged, checkSparkAnswerAndOperator +-- would detect differing struct values. This covers the "dual computation path" +-- cross-validation concern. + +-- ============================================================= +-- Setup: Create table with bucket partitioning (format v2 for MoR) +-- ============================================================= + +statement +DROP TABLE IF EXISTS test_cat.db.meta_test + +statement +CREATE TABLE test_cat.db.meta_test (id INT, name STRING, value DOUBLE) USING iceberg PARTITIONED BY (bucket(4, id)) TBLPROPERTIES ('format-version' = '2') + +statement +INSERT INTO test_cat.db.meta_test VALUES (1, 'alice', 10.0), (2, 'bob', 20.0), (3, 'charlie', 30.0) + +statement +INSERT INTO test_cat.db.meta_test VALUES (4, 'dave', 40.0), (5, 'eve', 50.0) + +-- ============================================================= +-- _file: basic projection (native) +-- ============================================================= + +query +SELECT id, _file FROM test_cat.db.meta_test ORDER BY id + +-- Multiple inserts produce multiple files +query +SELECT COUNT(DISTINCT _file) > 1 FROM test_cat.db.meta_test + +-- _file in a WHERE clause exercises filter pushdown interaction +query +SELECT id, name FROM test_cat.db.meta_test WHERE _file LIKE '%.parquet' ORDER BY id + +-- ============================================================= +-- _partition: basic struct projection (native) +-- ============================================================= + +query +SELECT id, _partition FROM test_cat.db.meta_test ORDER BY id + +-- Access individual partition fields within the struct +query +SELECT id, _partition.id_bucket FROM test_cat.db.meta_test ORDER BY id + +-- ============================================================= +-- _spec_id: partition spec ID (scalar per file) +-- ============================================================= + +query +SELECT id, _spec_id FROM test_cat.db.meta_test ORDER BY id + +-- All rows should have spec_id = 0 (initial spec) +query +SELECT DISTINCT _spec_id FROM test_cat.db.meta_test + +-- ============================================================= +-- _pos: row position within each file (0-based) +-- ============================================================= + +-- Basic: positions should be 0-based within each file +query +SELECT id, _file, _pos FROM test_cat.db.meta_test ORDER BY _file, _pos + +-- Positions should be contiguous 0..N-1 per file +query +SELECT _file, MIN(_pos) as min_pos, MAX(_pos) as max_pos, COUNT(*) as cnt FROM test_cat.db.meta_test GROUP BY _file + +-- ============================================================= +-- Partition evolution requires IcebergSparkSessionExtensions +-- (ALTER TABLE ADD PARTITION FIELD / CALL system.add_partition_field). +-- The SQL file test framework cannot register session extensions via +-- Config directives. Partition evolution is covered in +-- CometIcebergNativeSuite which has full session control. +-- ============================================================= + +-- ============================================================= +-- Combined: all metadata columns together (native -- all supported) +-- ============================================================= + +query +SELECT id, _file, _pos, _spec_id, _partition FROM test_cat.db.meta_test ORDER BY _file, _pos + +-- Row-level operation simulation: what CoW/MoR would project +query +SELECT _file, _pos, _spec_id, _partition, id, name FROM test_cat.db.meta_test WHERE id > 3 ORDER BY _file, _pos + +-- ============================================================= +-- _file + _partition combined (native -- both supported) +-- ============================================================= + +query +SELECT id, _file, _partition FROM test_cat.db.meta_test ORDER BY id + +statement +DROP TABLE test_cat.db.meta_test + +-- ============================================================= +-- Identity partition transform +-- ============================================================= + +statement +DROP TABLE IF EXISTS test_cat.db.identity_part + +statement +CREATE TABLE test_cat.db.identity_part (id INT, name STRING, dept STRING) USING iceberg PARTITIONED BY (dept) + +statement +INSERT INTO test_cat.db.identity_part VALUES (1, 'Alice', 'eng'), (2, 'Bob', 'sales'), (3, 'Charlie', 'eng') + +query +SELECT id, _partition FROM test_cat.db.identity_part ORDER BY id + +query +SELECT id, _partition.dept FROM test_cat.db.identity_part ORDER BY id + +query +SELECT id, _file, _partition FROM test_cat.db.identity_part ORDER BY id + +statement +DROP TABLE test_cat.db.identity_part + +-- ============================================================= +-- Multiple partition fields +-- ============================================================= + +statement +DROP TABLE IF EXISTS test_cat.db.multi_part + +statement +CREATE TABLE test_cat.db.multi_part (id INT, year INT, month INT, data STRING) USING iceberg PARTITIONED BY (year, month) + +statement +INSERT INTO test_cat.db.multi_part VALUES (1, 2023, 6, 'a'), (2, 2023, 7, 'b'), (3, 2024, 1, 'c') + +query +SELECT id, _partition FROM test_cat.db.multi_part ORDER BY id + +statement +DROP TABLE test_cat.db.multi_part + +-- ============================================================= +-- Partition with filter +-- ============================================================= + +statement +DROP TABLE IF EXISTS test_cat.db.filter_part + +statement +CREATE TABLE test_cat.db.filter_part (id INT, dept STRING, value DOUBLE) USING iceberg PARTITIONED BY (dept) + +statement +INSERT INTO test_cat.db.filter_part VALUES (1, 'eng', 10.5), (2, 'sales', 20.3), (3, 'eng', 30.7) + +query +SELECT id, _partition FROM test_cat.db.filter_part WHERE dept = 'eng' ORDER BY id + +statement +DROP TABLE test_cat.db.filter_part + +-- ============================================================= +-- Edge case: Unpartitioned table +-- ============================================================= + +statement +DROP TABLE IF EXISTS test_cat.db.unpart + +statement +CREATE TABLE test_cat.db.unpart (id INT, val STRING) USING iceberg + +statement +INSERT INTO test_cat.db.unpart VALUES (1, 'a'), (2, 'b') + +-- _partition on unpartitioned table is an empty struct (null), which Comet's +-- shuffle doesn't support. Verify correctness only, not native operator coverage. +query spark_answer_only +SELECT id, _partition FROM test_cat.db.unpart ORDER BY id + +-- Without ORDER BY (no shuffle), the native scan executes end-to-end +query +SELECT id, _partition FROM test_cat.db.unpart + +query +SELECT id, _file FROM test_cat.db.unpart ORDER BY id + +-- _spec_id on unpartitioned: should be 0 +-- (spark_answer_only because _partition is an empty struct that Comet shuffle rejects) +query spark_answer_only +SELECT id, _spec_id, _partition FROM test_cat.db.unpart ORDER BY id + +statement +DROP TABLE test_cat.db.unpart + +-- ============================================================= +-- Edge case: Large batch (tests multi-batch _pos correctness) +-- ============================================================= + +statement +DROP TABLE IF EXISTS test_cat.db.large_pos + +statement +CREATE TABLE test_cat.db.large_pos (id INT) USING iceberg + +statement +INSERT INTO test_cat.db.large_pos SELECT id FROM range(10000) + +query +SELECT id, _pos FROM test_cat.db.large_pos ORDER BY id LIMIT 5 + +statement +DROP TABLE test_cat.db.large_pos + +-- ============================================================= +-- Edge case: Multiple row groups (forces _pos offset tracking) +-- ============================================================= + +statement +DROP TABLE IF EXISTS test_cat.db.multi_rg + +statement +CREATE TABLE test_cat.db.multi_rg (id INT, data STRING) USING iceberg TBLPROPERTIES ('write.parquet.row-group-size-bytes' = '1024') + +statement +INSERT INTO test_cat.db.multi_rg SELECT id, REPEAT('x', 100) FROM range(1000) + +query +SELECT _pos, id FROM test_cat.db.multi_rg WHERE _pos IN (0, 1, 998, 999) ORDER BY _pos + +statement +DROP TABLE test_cat.db.multi_rg + +-- ============================================================= +-- DML operations (DELETE, UPDATE, MERGE) internally project metadata columns +-- (_file, _pos, _spec_id, _partition) to identify rows. All four are now +-- supported natively. The subsequent SELECT queries verify that reads after +-- DML return correct results through the native scan path (applying position +-- deletes, seeing updated data). +-- ============================================================= + +-- Copy-on-write DELETE +statement +DROP TABLE IF EXISTS test_cat.db.cow_delete + +statement +CREATE TABLE test_cat.db.cow_delete (id INT, name STRING, dept STRING) USING iceberg PARTITIONED BY (dept) TBLPROPERTIES ('write.delete.mode'='copy-on-write') + +statement +INSERT INTO test_cat.db.cow_delete VALUES (1, 'Alice', 'eng'), (2, 'Bob', 'sales'), (3, 'Charlie', 'eng') + +statement +DELETE FROM test_cat.db.cow_delete WHERE id = 2 + +query +SELECT id, _partition FROM test_cat.db.cow_delete ORDER BY id + +statement +DROP TABLE test_cat.db.cow_delete + +-- Merge-on-read DELETE (native read applies position deletes) +statement +DROP TABLE IF EXISTS test_cat.db.mor_delete + +statement +CREATE TABLE test_cat.db.mor_delete (id INT, name STRING, dept STRING) USING iceberg PARTITIONED BY (dept) TBLPROPERTIES ('write.delete.mode'='merge-on-read') + +statement +INSERT INTO test_cat.db.mor_delete VALUES (1, 'Alice', 'eng'), (2, 'Bob', 'sales'), (3, 'Charlie', 'eng') + +statement +DELETE FROM test_cat.db.mor_delete WHERE id = 2 + +query +SELECT id, _partition FROM test_cat.db.mor_delete ORDER BY id + +-- After MoR delete, _pos still references original file positions +query +SELECT id, _pos FROM test_cat.db.mor_delete ORDER BY id + +statement +DROP TABLE test_cat.db.mor_delete + +-- NOTE: Merge-on-read UPDATE is not tested here because UPDATE TABLE is not +-- supported in Spark 4.0. Covered in CometIcebergNativeSuite instead. + +-- ============================================================= +-- Temporal transforms: years() on DATE column +-- ============================================================= + +statement +DROP TABLE IF EXISTS test_cat.db.part_years_date + +statement +CREATE TABLE test_cat.db.part_years_date (id INT, event_date DATE, data STRING) USING iceberg PARTITIONED BY (years(event_date)) + +statement +INSERT INTO test_cat.db.part_years_date VALUES (1, DATE '2022-03-15', 'a'), (2, DATE '2023-07-20', 'b'), (3, DATE '2022-11-01', 'c') + +query +SELECT id, _partition FROM test_cat.db.part_years_date ORDER BY id + +query +SELECT id, _partition.event_date_year FROM test_cat.db.part_years_date ORDER BY id + +statement +DROP TABLE test_cat.db.part_years_date + +-- ============================================================= +-- Temporal transforms: months() on DATE column +-- ============================================================= + +statement +DROP TABLE IF EXISTS test_cat.db.part_months_date + +statement +CREATE TABLE test_cat.db.part_months_date (id INT, event_date DATE, data STRING) USING iceberg PARTITIONED BY (months(event_date)) + +statement +INSERT INTO test_cat.db.part_months_date VALUES (1, DATE '2023-01-10', 'jan'), (2, DATE '2023-02-15', 'feb'), (3, DATE '2023-01-25', 'jan2') + +query +SELECT id, _partition FROM test_cat.db.part_months_date ORDER BY id + +query +SELECT id, _partition.event_date_month FROM test_cat.db.part_months_date ORDER BY id + +statement +DROP TABLE test_cat.db.part_months_date + +-- ============================================================= +-- Temporal transforms: days() on DATE column +-- ============================================================= + +statement +DROP TABLE IF EXISTS test_cat.db.part_days_date + +statement +CREATE TABLE test_cat.db.part_days_date (id INT, event_date DATE, data STRING) USING iceberg PARTITIONED BY (days(event_date)) + +statement +INSERT INTO test_cat.db.part_days_date VALUES (1, DATE '2023-06-01', 'x'), (2, DATE '2023-06-02', 'y'), (3, DATE '2023-06-01', 'z') + +query +SELECT id, _partition FROM test_cat.db.part_days_date ORDER BY id + +query +SELECT id, _partition.event_date_day FROM test_cat.db.part_days_date ORDER BY id + +statement +DROP TABLE test_cat.db.part_days_date + +-- ============================================================= +-- Temporal transforms: hours() on TIMESTAMP column +-- ============================================================= + +statement +DROP TABLE IF EXISTS test_cat.db.part_hours_ts + +statement +CREATE TABLE test_cat.db.part_hours_ts (id INT, ts TIMESTAMP, data STRING) USING iceberg PARTITIONED BY (hours(ts)) + +statement +INSERT INTO test_cat.db.part_hours_ts VALUES (1, TIMESTAMP '2023-06-15 10:30:00', 'morning'), (2, TIMESTAMP '2023-06-15 14:45:00', 'afternoon'), (3, TIMESTAMP '2023-06-15 10:55:00', 'morning2') + +query +SELECT id, _partition FROM test_cat.db.part_hours_ts ORDER BY id + +query +SELECT id, _partition.ts_hour FROM test_cat.db.part_hours_ts ORDER BY id + +statement +DROP TABLE test_cat.db.part_hours_ts + +-- ============================================================= +-- Truncate transform: truncate(N, int_col) +-- ============================================================= + +statement +DROP TABLE IF EXISTS test_cat.db.part_trunc_int + +statement +CREATE TABLE test_cat.db.part_trunc_int (id INT, value INT, data STRING) USING iceberg PARTITIONED BY (truncate(10, value)) + +statement +INSERT INTO test_cat.db.part_trunc_int VALUES (1, 5, 'a'), (2, 15, 'b'), (3, 7, 'c'), (4, 23, 'd') + +query +SELECT id, _partition FROM test_cat.db.part_trunc_int ORDER BY id + +query +SELECT id, _partition.value_trunc FROM test_cat.db.part_trunc_int ORDER BY id + +statement +DROP TABLE test_cat.db.part_trunc_int + +-- ============================================================= +-- Truncate transform: truncate(N, string_col) +-- ============================================================= + +statement +DROP TABLE IF EXISTS test_cat.db.part_trunc_str + +statement +CREATE TABLE test_cat.db.part_trunc_str (id INT, city STRING, data STRING) USING iceberg PARTITIONED BY (truncate(3, city)) + +statement +INSERT INTO test_cat.db.part_trunc_str VALUES (1, 'San Francisco', 'x'), (2, 'Santa Cruz', 'y'), (3, 'Boston', 'z') + +query +SELECT id, _partition FROM test_cat.db.part_trunc_str ORDER BY id + +query +SELECT id, _partition.city_trunc FROM test_cat.db.part_trunc_str ORDER BY id + +statement +DROP TABLE test_cat.db.part_trunc_str + +-- ============================================================= +-- Multiple non-identity transforms on same table +-- ============================================================= + +statement +DROP TABLE IF EXISTS test_cat.db.part_multi_transform + +statement +CREATE TABLE test_cat.db.part_multi_transform (id INT, name STRING, ts TIMESTAMP) USING iceberg PARTITIONED BY (bucket(4, id), truncate(2, name), hours(ts)) + +statement +INSERT INTO test_cat.db.part_multi_transform VALUES (1, 'Alice', TIMESTAMP '2023-06-15 10:00:00'), (2, 'Bob', TIMESTAMP '2023-06-15 11:00:00'), (3, 'Alice', TIMESTAMP '2023-06-15 10:30:00') + +query +SELECT id, _partition FROM test_cat.db.part_multi_transform ORDER BY id + +query +SELECT id, _partition.id_bucket, _partition.name_trunc, _partition.ts_hour FROM test_cat.db.part_multi_transform ORDER BY id + +statement +DROP TABLE test_cat.db.part_multi_transform + +-- ============================================================= +-- NULL partition values +-- ============================================================= + +statement +DROP TABLE IF EXISTS test_cat.db.part_nulls + +statement +CREATE TABLE test_cat.db.part_nulls (id INT, dept STRING, value DOUBLE) USING iceberg PARTITIONED BY (dept) + +statement +INSERT INTO test_cat.db.part_nulls VALUES (1, 'eng', 10.0), (2, NULL, 20.0), (3, 'sales', 30.0), (4, NULL, 40.0) + +query +SELECT id, _partition FROM test_cat.db.part_nulls ORDER BY id + +query +SELECT id, _partition.dept FROM test_cat.db.part_nulls ORDER BY id + +query +SELECT id, _partition FROM test_cat.db.part_nulls WHERE _partition.dept IS NULL ORDER BY id + +query +SELECT id, _partition FROM test_cat.db.part_nulls WHERE _partition.dept IS NOT NULL ORDER BY id + +statement +DROP TABLE test_cat.db.part_nulls + +-- ============================================================= +-- _partition in WHERE clause (filter on metadata column itself) +-- ============================================================= + +statement +DROP TABLE IF EXISTS test_cat.db.part_where + +statement +CREATE TABLE test_cat.db.part_where (id INT, dept STRING, value DOUBLE) USING iceberg PARTITIONED BY (dept) + +statement +INSERT INTO test_cat.db.part_where VALUES (1, 'eng', 10.0), (2, 'sales', 20.0), (3, 'eng', 30.0), (4, 'hr', 40.0) + +query +SELECT id, value FROM test_cat.db.part_where WHERE _partition.dept = 'eng' ORDER BY id + +query +SELECT id, _partition FROM test_cat.db.part_where WHERE _partition.dept IN ('eng', 'hr') ORDER BY id + +statement +DROP TABLE test_cat.db.part_where + +-- ============================================================= +-- _partition in GROUP BY +-- ============================================================= + +statement +DROP TABLE IF EXISTS test_cat.db.part_groupby + +statement +CREATE TABLE test_cat.db.part_groupby (id INT, dept STRING, salary DOUBLE) USING iceberg PARTITIONED BY (dept) + +statement +INSERT INTO test_cat.db.part_groupby VALUES (1, 'eng', 100.0), (2, 'eng', 120.0), (3, 'sales', 80.0), (4, 'sales', 90.0), (5, 'hr', 70.0) + +query +SELECT _partition.dept, COUNT(*) as cnt, SUM(salary) as total FROM test_cat.db.part_groupby GROUP BY _partition.dept ORDER BY dept + +-- ORDER BY on _partition (StructType) falls back because Comet does not support Sort on structs +query spark_answer_only +SELECT _partition, COUNT(*) as cnt FROM test_cat.db.part_groupby GROUP BY _partition ORDER BY _partition + +statement +DROP TABLE test_cat.db.part_groupby + +-- ============================================================= +-- _partition in JOIN condition +-- ============================================================= + +statement +DROP TABLE IF EXISTS test_cat.db.part_join_left + +statement +DROP TABLE IF EXISTS test_cat.db.part_join_right + +statement +CREATE TABLE test_cat.db.part_join_left (id INT, dept STRING, value INT) USING iceberg PARTITIONED BY (dept) + +statement +CREATE TABLE test_cat.db.part_join_right (id INT, dept STRING, budget INT) USING iceberg PARTITIONED BY (dept) + +statement +INSERT INTO test_cat.db.part_join_left VALUES (1, 'eng', 10), (2, 'sales', 20), (3, 'eng', 30) + +statement +INSERT INTO test_cat.db.part_join_right VALUES (100, 'eng', 500), (200, 'sales', 300) + +query +SELECT l.id, l.value, r.budget FROM test_cat.db.part_join_left l JOIN test_cat.db.part_join_right r ON l._partition.dept = r._partition.dept ORDER BY l.id + +statement +DROP TABLE test_cat.db.part_join_left + +statement +DROP TABLE test_cat.db.part_join_right + +-- ============================================================= +-- _partition in subquery +-- ============================================================= + +statement +DROP TABLE IF EXISTS test_cat.db.part_subq + +statement +CREATE TABLE test_cat.db.part_subq (id INT, dept STRING, value INT) USING iceberg PARTITIONED BY (dept) + +statement +INSERT INTO test_cat.db.part_subq VALUES (1, 'eng', 10), (2, 'sales', 20), (3, 'hr', 30), (4, 'eng', 40) + +query +SELECT id, value FROM test_cat.db.part_subq WHERE _partition.dept IN (SELECT _partition.dept FROM test_cat.db.part_subq WHERE value > 25) ORDER BY id + +statement +DROP TABLE test_cat.db.part_subq + +-- ============================================================= +-- _partition with HAVING clause +-- ============================================================= + +statement +DROP TABLE IF EXISTS test_cat.db.part_having + +statement +CREATE TABLE test_cat.db.part_having (id INT, dept STRING, salary DOUBLE) USING iceberg PARTITIONED BY (dept) + +statement +INSERT INTO test_cat.db.part_having VALUES (1, 'eng', 100.0), (2, 'eng', 120.0), (3, 'eng', 110.0), (4, 'sales', 80.0), (5, 'hr', 70.0) + +query +SELECT _partition.dept, COUNT(*) as cnt FROM test_cat.db.part_having GROUP BY _partition.dept HAVING COUNT(*) > 1 ORDER BY dept + +statement +DROP TABLE test_cat.db.part_having + +-- ============================================================= +-- _partition with ORDER BY on partition field +-- ============================================================= + +statement +DROP TABLE IF EXISTS test_cat.db.part_orderby + +statement +CREATE TABLE test_cat.db.part_orderby (id INT, dept STRING, priority INT) USING iceberg PARTITIONED BY (dept) + +statement +INSERT INTO test_cat.db.part_orderby VALUES (1, 'eng', 3), (2, 'sales', 1), (3, 'hr', 2), (4, 'eng', 4) + +query +SELECT id, _partition.dept FROM test_cat.db.part_orderby ORDER BY _partition.dept, id + +statement +DROP TABLE test_cat.db.part_orderby + +-- ============================================================= +-- _partition with DISTINCT +-- ============================================================= + +statement +DROP TABLE IF EXISTS test_cat.db.part_distinct + +statement +CREATE TABLE test_cat.db.part_distinct (id INT, region STRING, data STRING) USING iceberg PARTITIONED BY (region) + +statement +INSERT INTO test_cat.db.part_distinct VALUES (1, 'US', 'a'), (2, 'EU', 'b'), (3, 'US', 'c'), (4, 'EU', 'd'), (5, 'APAC', 'e') + +query spark_answer_only +SELECT DISTINCT _partition.region FROM test_cat.db.part_distinct ORDER BY region + +query spark_answer_only +SELECT DISTINCT _partition FROM test_cat.db.part_distinct ORDER BY _partition.region + +statement +DROP TABLE test_cat.db.part_distinct + +-- ============================================================= +-- Wide partition struct (many partition fields) +-- ============================================================= + +statement +DROP TABLE IF EXISTS test_cat.db.part_wide + +statement +CREATE TABLE test_cat.db.part_wide (id INT, a INT, b INT, c INT, d STRING, e STRING) USING iceberg PARTITIONED BY (a, b, c, d, e) + +statement +INSERT INTO test_cat.db.part_wide VALUES (1, 10, 20, 30, 'x', 'y'), (2, 11, 21, 31, 'p', 'q'), (3, 10, 20, 30, 'x', 'y') + +query +SELECT id, _partition FROM test_cat.db.part_wide ORDER BY id + +query +SELECT id, _partition.a, _partition.b, _partition.c, _partition.d, _partition.e FROM test_cat.db.part_wide ORDER BY id + +statement +DROP TABLE test_cat.db.part_wide + +-- ============================================================= +-- DATE identity partition +-- ============================================================= + +statement +DROP TABLE IF EXISTS test_cat.db.part_date_identity + +statement +CREATE TABLE test_cat.db.part_date_identity (id INT, event_date DATE, data STRING) USING iceberg PARTITIONED BY (event_date) + +statement +INSERT INTO test_cat.db.part_date_identity VALUES (1, DATE '2023-01-15', 'a'), (2, DATE '2023-02-20', 'b'), (3, DATE '2023-01-15', 'c') + +query +SELECT id, _partition FROM test_cat.db.part_date_identity ORDER BY id + +query +SELECT id, _partition.event_date FROM test_cat.db.part_date_identity ORDER BY id + +query +SELECT id, data FROM test_cat.db.part_date_identity WHERE _partition.event_date = DATE '2023-01-15' ORDER BY id + +statement +DROP TABLE test_cat.db.part_date_identity + +-- ============================================================= +-- String identity with special characters +-- ============================================================= + +statement +DROP TABLE IF EXISTS test_cat.db.part_special_chars + +statement +CREATE TABLE test_cat.db.part_special_chars (id INT, tag STRING, data STRING) USING iceberg PARTITIONED BY (tag) + +statement +INSERT INTO test_cat.db.part_special_chars VALUES (1, 'hello world', 'a'), (2, 'foo/bar', 'b'), (3, '', 'c'), (4, 'a=b&c=d', 'd') + +query +SELECT id, _partition FROM test_cat.db.part_special_chars ORDER BY id + +query +SELECT id, _partition.tag FROM test_cat.db.part_special_chars ORDER BY id + +query +SELECT id FROM test_cat.db.part_special_chars WHERE _partition.tag = 'hello world' ORDER BY id + +query +SELECT id FROM test_cat.db.part_special_chars WHERE _partition.tag = '' ORDER BY id + +statement +DROP TABLE test_cat.db.part_special_chars + +-- ============================================================= +-- Bucket on STRING column +-- ============================================================= + +statement +DROP TABLE IF EXISTS test_cat.db.part_bucket_str + +statement +CREATE TABLE test_cat.db.part_bucket_str (id INT, name STRING, value INT) USING iceberg PARTITIONED BY (bucket(8, name)) + +statement +INSERT INTO test_cat.db.part_bucket_str VALUES (1, 'Alice', 10), (2, 'Bob', 20), (3, 'Charlie', 30), (4, 'Alice', 40) + +query +SELECT id, _partition FROM test_cat.db.part_bucket_str ORDER BY id + +query +SELECT id, _partition.name_bucket FROM test_cat.db.part_bucket_str ORDER BY id + +statement +DROP TABLE test_cat.db.part_bucket_str + +-- ============================================================= +-- years() on TIMESTAMP column +-- ============================================================= + +statement +DROP TABLE IF EXISTS test_cat.db.part_years_ts + +statement +CREATE TABLE test_cat.db.part_years_ts (id INT, ts TIMESTAMP, data STRING) USING iceberg PARTITIONED BY (years(ts)) + +statement +INSERT INTO test_cat.db.part_years_ts VALUES (1, TIMESTAMP '2021-03-15 08:00:00', 'a'), (2, TIMESTAMP '2022-07-20 12:00:00', 'b'), (3, TIMESTAMP '2021-11-01 16:00:00', 'c') + +query +SELECT id, _partition FROM test_cat.db.part_years_ts ORDER BY id + +query +SELECT id, _partition.ts_year FROM test_cat.db.part_years_ts ORDER BY id + +statement +DROP TABLE test_cat.db.part_years_ts + +-- ============================================================= +-- months() on TIMESTAMP column +-- ============================================================= + +statement +DROP TABLE IF EXISTS test_cat.db.part_months_ts + +statement +CREATE TABLE test_cat.db.part_months_ts (id INT, ts TIMESTAMP, data STRING) USING iceberg PARTITIONED BY (months(ts)) + +statement +INSERT INTO test_cat.db.part_months_ts VALUES (1, TIMESTAMP '2023-01-10 09:00:00', 'jan'), (2, TIMESTAMP '2023-02-15 10:00:00', 'feb'), (3, TIMESTAMP '2023-01-25 11:00:00', 'jan2') + +query +SELECT id, _partition FROM test_cat.db.part_months_ts ORDER BY id + +query +SELECT id, _partition.ts_month FROM test_cat.db.part_months_ts ORDER BY id + +statement +DROP TABLE test_cat.db.part_months_ts + +-- ============================================================= +-- Combined: temporal transform + identity on same table +-- ============================================================= + +statement +DROP TABLE IF EXISTS test_cat.db.part_mixed + +statement +CREATE TABLE test_cat.db.part_mixed (id INT, event_date DATE, region STRING, data STRING) USING iceberg PARTITIONED BY (years(event_date), region) + +statement +INSERT INTO test_cat.db.part_mixed VALUES (1, DATE '2022-03-01', 'US', 'a'), (2, DATE '2023-07-15', 'EU', 'b'), (3, DATE '2022-09-20', 'US', 'c'), (4, DATE '2023-01-10', 'US', 'd') + +query +SELECT id, _partition FROM test_cat.db.part_mixed ORDER BY id + +query +SELECT id, _partition.event_date_year, _partition.region FROM test_cat.db.part_mixed ORDER BY id + +query +SELECT _partition.event_date_year, COUNT(*) as cnt FROM test_cat.db.part_mixed GROUP BY _partition.event_date_year ORDER BY event_date_year + +query +SELECT id, data FROM test_cat.db.part_mixed WHERE _partition.region = 'US' ORDER BY id + +statement +DROP TABLE test_cat.db.part_mixed + +-- ============================================================= +-- _partition after multiple inserts (different files, same partition) +-- ============================================================= + +statement +DROP TABLE IF EXISTS test_cat.db.part_multi_insert + +statement +CREATE TABLE test_cat.db.part_multi_insert (id INT, dept STRING, value INT) USING iceberg PARTITIONED BY (dept) + +statement +INSERT INTO test_cat.db.part_multi_insert VALUES (1, 'eng', 10), (2, 'sales', 20) + +statement +INSERT INTO test_cat.db.part_multi_insert VALUES (3, 'eng', 30), (4, 'hr', 40) + +statement +INSERT INTO test_cat.db.part_multi_insert VALUES (5, 'sales', 50) + +query +SELECT id, _partition FROM test_cat.db.part_multi_insert ORDER BY id + +query +SELECT _partition.dept, COUNT(DISTINCT _file) as file_cnt, COUNT(*) as row_cnt FROM test_cat.db.part_multi_insert GROUP BY _partition.dept ORDER BY dept + +statement +DROP TABLE test_cat.db.part_multi_insert diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala index bdad3163d4..9a20a70a13 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala @@ -4806,8 +4806,9 @@ class CometIcebergNativeSuite } } - test("CometScanRule should report unsupported metadata columns") { + test("metadata columns - _pos returns row position within each file") { assume(icebergAvailable, "Iceberg not available in classpath") + withTempIcebergDir { warehouseDir => withSQLConf( "spark.sql.catalog.test_cat" -> "org.apache.iceberg.spark.SparkCatalog", @@ -4815,24 +4816,251 @@ class CometIcebergNativeSuite "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", - CometConf.COMET_SCALA_UDF_CODEGEN_ENABLED.key -> "true") { + CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") { - val table = "test_cat.db.test_meta_cols" + val table = "test_cat.db.pos_test" + try { + spark.sql(s""" + CREATE TABLE $table (id INT, name STRING) USING iceberg + """) + + // coalesce(1) guarantees a single file so _pos is 0,1,2 + spark + .sql("SELECT 1 as id, 'Alice' as name UNION ALL SELECT 2, 'Bob' UNION ALL SELECT 3, 'Charlie'") + .coalesce(1) + .write + .format("iceberg") + .mode("append") + .saveAsTable(table) + + checkIcebergNativeScan(s"SELECT id, _pos FROM $table ORDER BY id") + + // _pos should be 0-indexed within the single file + val positions = spark + .sql(s"SELECT _pos FROM $table ORDER BY _pos") + .collect() + .map(_.getLong(0)) + assert(positions.toSeq == Seq(0L, 1L, 2L), s"Expected [0,1,2], got ${positions.toSeq}") + } finally { + spark.sql(s"DROP TABLE IF EXISTS $table PURGE") + } + } + } + } + + test("metadata columns - _pos resets per file with multiple data files") { + 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.pos_multi_file_test" try { spark.sql(s""" CREATE TABLE $table (id INT, value DOUBLE) USING iceberg + """) + + // Two coalesced inserts guarantee exactly two data files + spark + .range(1, 4) + .selectExpr("CAST(id AS INT)", "CAST(id * 10.0 AS DOUBLE) as value") + .coalesce(1) + .write + .format("iceberg") + .mode("append") + .saveAsTable(table) + spark + .range(4, 6) + .selectExpr("CAST(id AS INT)", "CAST(id * 10.0 AS DOUBLE) as value") + .coalesce(1) + .write + .format("iceberg") + .mode("append") + .saveAsTable(table) + + // _pos resets to 0 at the start of each file + checkIcebergNativeScan(s"SELECT id, _pos, _file FROM $table ORDER BY _file, _pos") + + // Verify each file starts _pos at 0 + val rows = spark + .sql(s"SELECT _file, _pos FROM $table ORDER BY _file, _pos") + .collect() + val byFile = rows.groupBy(_.getString(0)) + assert(byFile.size == 2, s"Expected 2 files, got ${byFile.size}") + byFile.foreach { case (file, fileRows) => + val positions = fileRows.map(_.getLong(1)) + assert( + positions.head == 0L, + s"File $file should start _pos at 0, got ${positions.head}") + } + } finally { + spark.sql(s"DROP TABLE IF EXISTS $table PURGE") + } + } + } + } + + test("metadata columns - _spec_id returns partition spec ID") { + 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.spec_id_test" + try { + spark.sql(s""" + CREATE TABLE $table (id INT, category STRING, value DOUBLE) + USING iceberg PARTITIONED BY (category) + """) + + spark.sql(s""" + INSERT INTO $table VALUES + (1, 'A', 10.0), (2, 'B', 20.0), (3, 'A', 30.0) + """) + + checkIcebergNativeScan(s"SELECT id, _spec_id FROM $table ORDER BY id") + + // All rows written under the initial spec should have _spec_id = 0 + val result = spark.sql(s"SELECT DISTINCT _spec_id FROM $table").collect() + assert(result.length == 1) + assert(result(0).getInt(0) == 0) + } finally { + spark.sql(s"DROP TABLE IF EXISTS $table PURGE") + } + } + } + } + + test("metadata columns - _spec_id changes after partition evolution") { + 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") { + + import org.apache.iceberg.catalog.TableIdentifier + import org.apache.iceberg.spark.SparkCatalog + + val table = "test_cat.db.spec_id_evolution" + try { + spark.sql(s""" + CREATE TABLE $table (id INT, region STRING, category STRING) + USING iceberg PARTITIONED BY (region) + """) + + // Data under spec 0 + spark.sql(s"INSERT INTO $table VALUES (1, 'US', 'A'), (2, 'EU', 'B')") + + // Evolve partition spec: add category field -> spec 1 + val sparkCatalog = spark.sessionState.catalogManager + .catalog("test_cat") + .asInstanceOf[SparkCatalog] + val iceTable = sparkCatalog + .icebergCatalog() + .loadTable(TableIdentifier.of("db", "spec_id_evolution")) + iceTable.updateSpec().addField("category").commit() + + // Data under spec 1 + spark.sql(s"INSERT INTO $table VALUES (3, 'APAC', 'C'), (4, 'US', 'D')") + + checkIcebergNativeScan(s"SELECT id, _spec_id FROM $table ORDER BY id") + + // Rows 1,2 should have spec_id=0; rows 3,4 should have spec_id=1 + val result = spark.sql(s"SELECT id, _spec_id FROM $table ORDER BY id").collect() + assert(result(0).getInt(1) == 0, "row 1 should have spec_id=0") + assert(result(1).getInt(1) == 0, "row 2 should have spec_id=0") + assert(result(2).getInt(1) == 1, "row 3 should have spec_id=1") + assert(result(3).getInt(1) == 1, "row 4 should have spec_id=1") + } finally { + spark.sql(s"DROP TABLE IF EXISTS $table PURGE") + } + } + } + } + + test("metadata columns - all metadata columns together") { + 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.all_meta_cols" + try { + spark.sql(s""" + CREATE TABLE $table (id INT, value DOUBLE) + USING iceberg PARTITIONED BY (bucket(4, id)) TBLPROPERTIES ('format-version' = '2') """) spark.sql(s""" - INSERT INTO $table - VALUES (1, 10.5), (2, 20.3), (3, 30.7) - """) + INSERT INTO $table VALUES (1, 10.5), (2, 20.3), (3, 30.7) + """) + + // Query all supported metadata columns together + checkIcebergNativeScan( + s"SELECT id, value, _file, _pos, _spec_id, _partition FROM $table ORDER BY id") + } finally { + spark.sql(s"DROP TABLE IF EXISTS $table PURGE") + } + } + } + } + + test("metadata columns - _pos with MOR deletes") { + 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") { - checkSparkAnswerAndFallbackReason( - s"SELECT id, value, _spec_id, _pos, _file, _partition FROM $table WHERE id >= 2 ORDER BY id", - "Metadata column(s) _spec_id, _partition, _file, _pos is not supported") + val table = "test_cat.db.pos_delete_test" + try { + spark.sql(s""" + CREATE TABLE $table (id INT, name STRING) USING iceberg + TBLPROPERTIES ( + 'write.delete.mode' = 'merge-on-read', + 'write.merge.mode' = 'merge-on-read' + ) + """) + + spark.sql(s""" + INSERT INTO $table VALUES + (1, 'Alice'), (2, 'Bob'), (3, 'Charlie'), (4, 'Diana'), (5, 'Eve') + """) + + // Delete some rows (creates position delete files) + spark.sql(s"DELETE FROM $table WHERE id IN (2, 4)") + + // _pos should still reflect original file positions for surviving rows + checkIcebergNativeScan(s"SELECT id, _pos FROM $table ORDER BY id") } finally { spark.sql(s"DROP TABLE IF EXISTS $table PURGE") } @@ -4870,7 +5098,6 @@ class CometIcebergNativeSuite test("variant column filter falls back to Spark") { assume(isSpark40Plus, "VARIANT type requires Spark 4.0+") assume(icebergAvailable, "Iceberg not available in classpath") - assume(icebergVersionAtLeast(1, 11), "VARIANT column type requires Iceberg 1.11+") withTempIcebergDir { warehouseDir => withSQLConf( "spark.sql.catalog.test_cat" -> "org.apache.iceberg.spark.SparkCatalog", @@ -4896,4 +5123,246 @@ class CometIcebergNativeSuite } } } + + test("partition evolution - _partition contains fields from all historical specs") { + 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") { + + import org.apache.iceberg.catalog.TableIdentifier + import org.apache.iceberg.spark.SparkCatalog + + spark.sql(""" + CREATE TABLE test_cat.db.part_evolution ( + id INT, region STRING, category STRING, value DOUBLE + ) USING iceberg PARTITIONED BY (region) + """) + + spark.sql(""" + INSERT INTO test_cat.db.part_evolution VALUES + (1, 'US', 'A', 10.0), (2, 'EU', 'B', 20.0) + """) + + // Add a second partition field via Iceberg Java API (partition evolution -> spec_id 1) + val sparkCatalog = spark.sessionState.catalogManager + .catalog("test_cat") + .asInstanceOf[SparkCatalog] + val table = sparkCatalog + .icebergCatalog() + .loadTable(TableIdentifier.of("db", "part_evolution")) + table.updateSpec().addField("category").commit() + + spark.sql(""" + INSERT INTO test_cat.db.part_evolution VALUES + (3, 'US', 'C', 30.0), (4, 'APAC', 'A', 40.0) + """) + + // _partition should be a struct with BOTH region and category fields, + // covering all historical specs. Files written under spec 0 will have + // category=null in _partition; files under spec 1 have both populated. + checkIcebergNativeScan( + "SELECT id, _partition FROM test_cat.db.part_evolution ORDER BY id") + + // Verify the struct fields are accessible + checkIcebergNativeScan( + "SELECT id, _partition.region, _partition.category " + + "FROM test_cat.db.part_evolution ORDER BY id") + + // Verify correctness: old rows have null category in _partition + val result = spark + .sql("SELECT id, _partition.category " + + "FROM test_cat.db.part_evolution ORDER BY id") + .collect() + assert(result(0).isNullAt(1), "row 1 (spec 0) should have null _partition.category") + assert(result(1).isNullAt(1), "row 2 (spec 0) should have null _partition.category") + assert(result(2).getString(1) == "C", "row 3 (spec 1) should have category=C") + assert(result(3).getString(1) == "A", "row 4 (spec 1) should have category=A") + + spark.sql("DROP TABLE test_cat.db.part_evolution") + } + } + } + + test("partition evolution - _partition struct fields sorted by field_id ascending") { + 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") { + + import org.apache.iceberg.catalog.TableIdentifier + import org.apache.iceberg.spark.SparkCatalog + + // Create table partitioned by region (gets lower field_id in partition spec) + spark.sql(""" + CREATE TABLE test_cat.db.field_order ( + id INT, region STRING, category STRING + ) USING iceberg PARTITIONED BY (region) + """) + + spark.sql(""" + INSERT INTO test_cat.db.field_order VALUES (1, 'US', 'A'), (2, 'EU', 'B') + """) + + // Evolve: add category as second partition field (gets higher field_id). + // Without sort-by-id, spec-descending traversal would emit category before region. + val sparkCatalog = spark.sessionState.catalogManager + .catalog("test_cat") + .asInstanceOf[SparkCatalog] + val table = sparkCatalog + .icebergCatalog() + .loadTable(TableIdentifier.of("db", "field_order")) + table.updateSpec().addField("category").commit() + + spark.sql(""" + INSERT INTO test_cat.db.field_order VALUES (3, 'APAC', 'C'), (4, 'US', 'D') + """) + + // Verify _partition struct field order matches Spark (sorted by field_id ascending). + // Spark's Java reader uses buildPartitionProjectionType which sorts by natural key order. + checkIcebergNativeScan("SELECT id, _partition FROM test_cat.db.field_order ORDER BY id") + + // Access fields by name to confirm both are present and correctly ordered + checkIcebergNativeScan( + "SELECT id, _partition.region, _partition.category " + + "FROM test_cat.db.field_order ORDER BY id") + + spark.sql("DROP TABLE test_cat.db.field_order") + } + } + } + + test("partition evolution - dropped partition field preserved in _partition struct") { + // Exercises the case CTTY raised in iceberg-rust PR #2668: + // Java's Partitioning.buildPartitionProjectionType preserves the type of a + // partition field that was dropped (Void in newer spec, real transform in older spec). + // This test verifies iceberg-rust matches that behavior through Comet. + 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") { + + import org.apache.iceberg.catalog.TableIdentifier + import org.apache.iceberg.spark.SparkCatalog + + val table = "test_cat.db.drop_part_field" + try { + // Spec 0: partitioned by (region, category) + spark.sql(s""" + CREATE TABLE $table (id INT, region STRING, category STRING, value DOUBLE) + USING iceberg PARTITIONED BY (region, category) + """) + + spark.sql(s"INSERT INTO $table VALUES (1, 'US', 'A', 10.0), (2, 'EU', 'B', 20.0)") + + // Evolve: drop 'category' from partition spec -> spec 1 marks category as Void + val sparkCatalog = spark.sessionState.catalogManager + .catalog("test_cat") + .asInstanceOf[SparkCatalog] + val iceTable = sparkCatalog + .icebergCatalog() + .loadTable(TableIdentifier.of("db", "drop_part_field")) + iceTable.updateSpec().removeField("category").commit() + + // Insert data under spec 1 (only region partitioning) + spark.sql(s"INSERT INTO $table VALUES (3, 'APAC', 'C', 30.0), (4, 'US', 'D', 40.0)") + + // _partition struct should still contain BOTH region and category fields. + // Java preserves dropped partition fields in the unified type; + // rows written under spec 1 have category=null in _partition. + checkIcebergNativeScan(s"SELECT id, _partition FROM $table ORDER BY id") + + checkIcebergNativeScan( + s"SELECT id, _partition.region, _partition.category FROM $table ORDER BY id") + + // Verify: spec-0 rows have category populated, spec-1 rows have category=null + val result = spark + .sql(s"SELECT id, _partition.category FROM $table ORDER BY id") + .collect() + assert( + result(0).getString(1) == "A", + "row 1 (spec 0) should have _partition.category=A") + assert( + result(1).getString(1) == "B", + "row 2 (spec 0) should have _partition.category=B") + assert(result(2).isNullAt(1), "row 3 (spec 1) should have null _partition.category") + assert(result(3).isNullAt(1), "row 4 (spec 1) should have null _partition.category") + } finally { + spark.sql(s"DROP TABLE IF EXISTS $table PURGE") + } + } + } + } + + test("partition evolution - re-add dropped partition field") { + // Further exercises the void-transform handling: drop a field then re-add it. + // Java's unified type should contain the field once (not duplicated). + 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") { + + import org.apache.iceberg.catalog.TableIdentifier + import org.apache.iceberg.spark.SparkCatalog + + val table = "test_cat.db.readd_part_field" + try { + // Spec 0: partitioned by (region) + spark.sql(s""" + CREATE TABLE $table (id INT, region STRING, value DOUBLE) + USING iceberg PARTITIONED BY (region) + """) + + spark.sql(s"INSERT INTO $table VALUES (1, 'US', 10.0)") + + val sparkCatalog = spark.sessionState.catalogManager + .catalog("test_cat") + .asInstanceOf[SparkCatalog] + val iceTable = sparkCatalog + .icebergCatalog() + .loadTable(TableIdentifier.of("db", "readd_part_field")) + + // Spec 1: drop region + iceTable.updateSpec().removeField("region").commit() + spark.sql(s"INSERT INTO $table VALUES (2, 'EU', 20.0)") + + // Spec 2: re-add region (gets a new partition field_id but same source column) + iceTable.refresh() + iceTable.updateSpec().addField("region").commit() + spark.sql(s"INSERT INTO $table VALUES (3, 'APAC', 30.0)") + + // All rows should be queryable with _partition + checkIcebergNativeScan(s"SELECT id, _partition FROM $table ORDER BY id") + + checkIcebergNativeScan(s"SELECT id, _partition.region FROM $table ORDER BY id") + } finally { + spark.sql(s"DROP TABLE IF EXISTS $table PURGE") + } + } + } + } } From e80b37dbccdb6db4e6716ee4633c9bf35bca970f Mon Sep 17 00:00:00 2001 From: Parth Chandra Date: Thu, 30 Jul 2026 18:50:43 -0700 Subject: [PATCH 2/6] address review comments --- native/core/src/execution/planner.rs | 178 +++++++++++++++-- native/proto/src/proto/operator.proto | 6 +- .../comet/iceberg/IcebergReflection.scala | 47 +++++ .../apache/comet/rules/CometScanRule.scala | 25 +-- .../operator/CometIcebergNativeScan.scala | 153 +++++++-------- .../iceberg/metadata_column_partition.sql | 8 +- .../comet/CometIcebergNativeSuite.scala | 179 +++++++++++++++++- 7 files changed, 465 insertions(+), 131 deletions(-) diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs index 116c11770f..61c8195846 100644 --- a/native/core/src/execution/planner.rs +++ b/native/core/src/execution/planner.rs @@ -3837,34 +3837,68 @@ fn parse_file_scan_tasks_from_common( // Compute unified partition type from the partition_type_pool. // Each entry is a StructType JSON with the resolved field types for one partition spec. - // Merge all specs into a single unified type (dedup by field_id). + // Merge all specs into a single unified type, matching Iceberg Java's + // Partitioning.buildPartitionProjectionType()/iceberg-rust's own + // compute_unified_partition_type(): process specs by spec_id descending (so the newest + // spec's field name wins when two specs share a field id), then sort the merged fields + // ascending by field id. The ascending sort matters beyond cosmetics: it fixes the + // physical Arrow struct field order for `_partition`, which must match the field order + // Spark's analyzer bound `_partition.` accesses to (also computed via + // Partitioning.partitionType(), spec-descending + ascending-sort) -- a mismatch would + // read the wrong value under the right field name instead of erroring. // - // NOTE: The type information here comes from Iceberg Java's PartitionSpec.partitionType() - // (via Scala reflection). This is the same type computation as iceberg-rust's - // Transform::result_type() -- both implement the Iceberg spec's transform result rules. - // When iceberg-rust's own scan planning is used (not Comet's proto path), it computes - // this via compute_unified_partition_type(specs, schema) instead. + // partition_type_pool is index-aligned with partition_spec_pool (see the .proto comment), + // so partition_spec_cache[i].spec_id() gives the spec_id for partition_type_pool[i]. + // + // NOTE: The resolved field types here come from Iceberg Java's PartitionSpec.partitionType() + // (via Scala reflection), which already applied Transform::result_type() per field. Because + // we only have resolved types (not the original Transform), we cannot replicate one further + // nuance of the reference algorithms: preferring a non-void transform's type over an older + // void transform's for the same field id. In practice this only affects a field whose + // partition source column was later dropped from the schema, which Comet already filters out + // upstream (see the "unknown type" filtering in serializePartitionData). let unified_partition_type = { + let mut indexed_types = proto_common + .partition_type_pool + .iter() + .enumerate() + .map(|(idx, type_json)| { + let struct_type = serde_json::from_str::(type_json) + .map_err(|e| { + ExecutionError::GeneralError(format!( + "Failed to deserialize partition type JSON from pool: {e}" + )) + })?; + let spec_id = partition_spec_cache + .get(idx) + .and_then(|opt| opt.as_ref()) + .map(|spec| spec.spec_id()) + .ok_or_else(|| { + ExecutionError::GeneralError(format!( + "No partition spec at pool index {idx} matching partition_type_pool \ + entry; partition_type_pool must be index-aligned with \ + partition_spec_pool" + )) + })?; + Ok((spec_id, struct_type)) + }) + .collect::, ExecutionError>>()?; + + indexed_types.sort_by_key(|(spec_id, _)| std::cmp::Reverse(*spec_id)); + let mut seen_field_ids = std::collections::HashSet::new(); let mut struct_fields: Vec = Vec::new(); - for type_json in &proto_common.partition_type_pool { - match serde_json::from_str::(type_json) { - Ok(struct_type) => { - for field in struct_type.fields() { - if seen_field_ids.insert(field.id) { - struct_fields.push(Arc::clone(field)); - } - } - } - Err(e) => { - return Err(ExecutionError::GeneralError(format!( - "Failed to deserialize partition type JSON from pool: {e}" - ))); + for (_, struct_type) in &indexed_types { + for field in struct_type.fields() { + if seen_field_ids.insert(field.id) { + struct_fields.push(Arc::clone(field)); } } } + struct_fields.sort_by_key(|f| f.id); + iceberg::spec::StructType::new(struct_fields) }; @@ -5516,4 +5550,110 @@ mod tests { "RESERVED_FIELD_ID_PARTITION must be i32::MAX - 5 to match Scala MetadataFieldIds" ); } + + #[test] + fn test_unified_partition_type_merges_specs_by_descending_spec_id() { + use iceberg::spec::{NestedField, PartitionSpec, PrimitiveType, Type}; + + let iceberg_schema = iceberg::spec::Schema::builder() + .with_schema_id(0) + .with_fields(vec![ + NestedField::required(1, "id", Type::Primitive(PrimitiveType::Int)).into(), + NestedField::required(2, "region", Type::Primitive(PrimitiveType::String)).into(), + NestedField::required(3, "category", Type::Primitive(PrimitiveType::String)).into(), + ]) + .build() + .expect("schema"); + let schema_arc = Arc::new(iceberg_schema); + + // Spec 0 (older): field 1000 named "region_old". + let spec0 = PartitionSpec::builder(Arc::clone(&schema_arc)) + .with_spec_id(0) + .add_unbound_field(iceberg::spec::UnboundPartitionField { + source_id: 2, + field_id: Some(1000), + name: "region_old".to_string(), + transform: iceberg::spec::Transform::Identity, + }) + .expect("add field") + .build() + .expect("build spec0"); + + // Spec 1 (newer): same field id 1000 renamed to "region_new", plus a new field 2000. + let spec1 = PartitionSpec::builder(Arc::clone(&schema_arc)) + .with_spec_id(1) + .add_unbound_field(iceberg::spec::UnboundPartitionField { + source_id: 2, + field_id: Some(1000), + name: "region_new".to_string(), + transform: iceberg::spec::Transform::Identity, + }) + .expect("add field") + .add_unbound_field(iceberg::spec::UnboundPartitionField { + source_id: 3, + field_id: Some(2000), + name: "category".to_string(), + transform: iceberg::spec::Transform::Identity, + }) + .expect("add field") + .build() + .expect("build spec1"); + + let spec0_type_json = + serde_json::to_string(&spec0.partition_type(&schema_arc).expect("partition_type")) + .expect("serialize"); + let spec1_type_json = + serde_json::to_string(&spec1.partition_type(&schema_arc).expect("partition_type")) + .expect("serialize"); + let spec0_json = serde_json::to_string(&spec0).expect("serialize spec0"); + let spec1_json = serde_json::to_string(&spec1).expect("serialize spec1"); + + let schema_json = serde_json::to_string(schema_arc.as_ref()).expect("serialize schema"); + + // Pool insertion order deliberately does NOT match spec_id order: the newer spec + // (spec_id 1) is inserted first, at index 0, and the older spec (spec_id 0) second, at + // index 1. If the merge relied on pool-insertion order instead of spec_id, this would + // produce the wrong result (region_old kept instead of region_new). + let proto_common = spark_operator::IcebergScanCommon { + schema_pool: vec![schema_json], + partition_type_pool: vec![spec1_type_json, spec0_type_json], + partition_spec_pool: vec![spec1_json, spec0_json], + project_field_ids_pool: vec![spark_operator::ProjectFieldIdList { + field_ids: vec![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), + project_field_ids_idx: 0, + ..Default::default() + }; + + let tasks = + parse_file_scan_tasks_from_common(&proto_common, &[proto_task]).expect("parse tasks"); + assert_eq!(tasks.len(), 1); + + let unified = tasks[0] + .unified_partition_type + .as_ref() + .expect("unified_partition_type must be set when _partition is projected"); + + let fields = unified.fields(); + assert_eq!( + fields.len(), + 2, + "expected fields 1000 and 2000, got {fields:?}" + ); + // Ascending by field id: 1000 before 2000. + assert_eq!(fields[0].id, 1000); + assert_eq!(fields[1].id, 2000); + // Newest spec (spec_id 1) wins the name for field 1000, despite being inserted into the + // pool before spec_id 0's entry. + assert_eq!(fields[0].name, "region_new"); + assert_eq!(fields[1].name, "category"); + } } diff --git a/native/proto/src/proto/operator.proto b/native/proto/src/proto/operator.proto index 1633429986..791db3d5ed 100644 --- a/native/proto/src/proto/operator.proto +++ b/native/proto/src/proto/operator.proto @@ -261,6 +261,11 @@ message IcebergScanCommon { // Deduplication pools (must contain all entries for cross-partition deduplication) repeated string schema_pool = 4; + // Index-aligned with partition_spec_pool: entry i is the StructType JSON for the spec at + // partition_spec_pool[i]. Native uses this pairing to recover each entry's spec_id when + // merging partition types across historical specs (see parse_file_scan_tasks_from_common + // in planner.rs), which partition_type_pool cannot express on its own since it is only + // resolved field types, not specs. repeated string partition_type_pool = 5; repeated string partition_spec_pool = 6; repeated string name_mapping_pool = 7; @@ -326,7 +331,6 @@ message IcebergFileScanTask { // Indices into IcebergScan deduplication pools uint32 schema_idx = 15; - optional uint32 partition_type_idx = 16; optional uint32 partition_spec_idx = 17; optional uint32 name_mapping_idx = 18; uint32 project_field_ids_idx = 19; 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 c06b1b4e47..7e8ba9df88 100644 --- a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala +++ b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala @@ -68,6 +68,36 @@ object IcebergReflection extends Logging { def isIcebergScanClass(name: String): Boolean = ICEBERG_SCAN_CLASSES.contains(name) + // Iceberg FileIO implementations whose underlying storage object_store can reach. + // Custom/test FileIO classes (e.g. CustomFileIO in TestSparkExecutorCache) are not compatible + // because Comet's native reader bypasses Java FileIO entirely. + val COMPATIBLE_FILE_IO_CLASSES: Set[String] = Set( + "org.apache.iceberg.hadoop.HadoopFileIO", + "org.apache.iceberg.aws.s3.S3FileIO", + "org.apache.iceberg.gcp.gcs.GCSFileIO", + "org.apache.iceberg.io.ResolvingFileIO", + "org.apache.iceberg.spark.SparkFileIO", + "org.apache.iceberg.azure.adlsv2.ADLSFileIO", + "org.apache.iceberg.CachingFileIO") + + // Prefix of the EncryptingFileIO family. An encrypted table's io() is not the bare + // EncryptingFileIO but a nested variant chosen from the wrapped delegate's capabilities + // (e.g. EncryptingFileIO$WithSupportsPrefixOperations when the delegate is HadoopFileIO), so + // an exact class-name match misses it. Comet forwards each file's key_metadata to iceberg-rust + // and reads the ciphertext via object_store, so any EncryptingFileIO variant is compatible. + private val ENCRYPTING_FILE_IO_PREFIX = "org.apache.iceberg.encryption.EncryptingFileIO" + + /** + * True if `fileIO` is a FileIO whose backing storage Comet's native reader can reach. Matches + * on `fileIO`'s own class hierarchy (via [[classNameInHierarchy]]) rather than its exact leaf + * class, so a subclass that only adds metrics/retry/credential-routing on top of a known- + * compatible FileIO (e.g. a custom S3FileIO subclass) still matches, instead of silently + * falling back to Spark. + */ + def isCompatibleFileIO(fileIO: Any): Boolean = + classNameInHierarchy(fileIO.getClass, COMPATIBLE_FILE_IO_CLASSES) || + fileIO.getClass.getName.startsWith(ENCRYPTING_FILE_IO_PREFIX) + /** * Iceberg content types. */ @@ -127,6 +157,23 @@ object IcebergReflection extends Logging { None } + /** + * True if `clazz` or any of its superclasses has a name in `names`. Walks the already-loaded + * class object's own hierarchy, so unlike [[loadClass]] it never risks a + * `ClassNotFoundException` for a candidate name that isn't on this JVM's classpath (e.g. + * checking for a GCS/Azure FileIO class when only iceberg-aws is bundled). + */ + def classNameInHierarchy(clazz: Class[_], names: Set[String]): Boolean = { + var current: Class[_] = clazz + while (current != null) { + if (names.contains(current.getName)) { + return true + } + current = current.getSuperclass + } + false + } + /** * Extracts file location from Iceberg ContentFile, handling both location() and path(). * 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 b936a91c18..41bc996052 100644 --- a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala +++ b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala @@ -477,7 +477,7 @@ case class CometScanRule(session: SparkSession) // FileIO entirely. Only allow known-compatible implementations whose underlying // storage object_store can reach via standard URL schemes. val fileIOCompatible = IcebergReflection.getFileIO(metadata.table) match { - case Some(fileIO) if CometScanRule.isCompatibleFileIO(fileIO.getClass.getName) => + case Some(fileIO) if IcebergReflection.isCompatibleFileIO(fileIO) => true case Some(fileIO) => fallbackReasons += s"FileIO ${fileIO.getClass.getName} is not supported by " + @@ -931,29 +931,6 @@ case class CometScanTypeChecker() extends DataTypeSupport with CometTypeShim { object CometScanRule extends Logging { - // Iceberg FileIO implementations whose underlying storage object_store can reach. - // Custom/test FileIO classes (e.g. CustomFileIO in TestSparkExecutorCache) are not compatible - // because Comet's native reader bypasses Java FileIO entirely. - private val CompatibleFileIOClasses: Set[String] = Set( - "org.apache.iceberg.hadoop.HadoopFileIO", - "org.apache.iceberg.aws.s3.S3FileIO", - "org.apache.iceberg.gcp.gcs.GCSFileIO", - "org.apache.iceberg.io.ResolvingFileIO", - "org.apache.iceberg.spark.SparkFileIO", - "org.apache.iceberg.azure.adlsv2.ADLSFileIO", - "org.apache.iceberg.CachingFileIO") - - // Prefix of the EncryptingFileIO family. An encrypted table's io() is not the bare - // EncryptingFileIO but a nested variant chosen from the wrapped delegate's capabilities - // (e.g. EncryptingFileIO$WithSupportsPrefixOperations when the delegate is HadoopFileIO), so - // an exact class-name match misses it. Comet forwards each file's key_metadata to iceberg-rust - // and reads the ciphertext via object_store, so any EncryptingFileIO variant is compatible. - private val EncryptingFileIOPrefix = "org.apache.iceberg.encryption.EncryptingFileIO" - - /** True if `className` is a FileIO whose backing storage Comet's native reader can reach. */ - def isCompatibleFileIO(className: String): Boolean = - CompatibleFileIOClasses.contains(className) || className.startsWith(EncryptingFileIOPrefix) - // Per-scheme memo of `NativeBase.isObjectStoreSchemeSupported`. The answer depends only on the // URL scheme, so we cache by scheme and never re-cross the JNI boundary for a repeated scheme. private val schemeSupportCache = 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 42ef14e9fa..6c4f45ce62 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 @@ -357,7 +357,6 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit fileScanTaskClass: Class[_], taskBuilder: OperatorOuterClass.IcebergFileScanTask.Builder, commonBuilder: OperatorOuterClass.IcebergScanCommon.Builder, - partitionTypeToPoolIndex: mutable.HashMap[String, Int], partitionSpecToPoolIndex: mutable.HashMap[String, Int], partitionDataToPoolIndex: mutable.HashMap[String, Int]): Unit = { try { @@ -365,7 +364,58 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit val spec = specMethod.invoke(task) if (spec != null) { - // Deduplicate partition spec + // Get the partition type/schema from the spec. Needed regardless of whether this task + // ends up value-less below. + val partitionTypeMethod = spec.getClass.getMethod("partitionType") + val partitionType = partitionTypeMethod.invoke(spec) + val fieldsMethod = partitionType.getClass.getMethod("fields") + val fields = fieldsMethod + .invoke(partitionType) + .asInstanceOf[java.util.List[_]] + + // Helper to get field type string (shared by both type and data serialization) + def getFieldType(field: Any): String = { + val typeMethod = field.getClass.getMethod("type") + typeMethod.invoke(field).toString + } + + // Filter out fields with unknown types (dropped partition fields). + // Unknown type fields represent partition columns that have been dropped + // from the schema. Per the Iceberg spec, unknown type fields are not + // stored in data files and iceberg-rust doesn't support deserializing + // them. Since these columns are dropped, we don't need to expose their + // partition values when reading. + val fieldsJson = fields.asScala.flatMap { field => + val fieldTypeStr = getFieldType(field) + + // Skip fields with unknown type (dropped partition columns) + if (fieldTypeStr == IcebergReflection.TypeNames.UNKNOWN) { + None + } else { + val fieldIdMethod = field.getClass.getMethod("fieldId") + val fieldId = fieldIdMethod.invoke(field).asInstanceOf[Int] + + val nameMethod = field.getClass.getMethod("name") + val fieldName = nameMethod.invoke(field).asInstanceOf[String] + + val isOptionalMethod = field.getClass.getMethod("isOptional") + val isOptional = + isOptionalMethod.invoke(field).asInstanceOf[Boolean] + val required = !isOptional + + Some( + ("id" -> fieldId) ~ + ("name" -> fieldName) ~ + ("required" -> required) ~ + ("type" -> fieldTypeStr)) + } + }.toList + + // Deduplicate partition spec, pairing each new spec-pool entry with a partition-type-pool + // entry at the same index. Native correlates the two by index to recover, per spec, the + // spec_id needed to merge partition types across historical specs in the same order + // Iceberg Java's Partitioning.partitionType()/iceberg-rust's compute_unified_partition_type + // do (newest spec_id first); see parse_file_scan_tasks_from_common in planner.rs. try { val partitionSpecParserClass = IcebergReflection.loadClass(IcebergReflection.ClassNames.PARTITION_SPEC_PARSER) @@ -380,6 +430,15 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit partitionSpecJson, { val idx = partitionSpecToPoolIndex.size commonBuilder.addPartitionSpecPool(partitionSpecJson) + // Manually build StructType JSON to match iceberg-rust expectations. + // Using Iceberg's SchemaParser.toJson() would include schema-level + // metadata (e.g., "schema-id") that iceberg-rust's StructType + // deserializer rejects. We need pure StructType format: + // {"type":"struct","fields":[...]} + val partitionTypeJson = compact( + render(("type" -> "struct") ~ + ("fields" -> fieldsJson))) + commonBuilder.addPartitionTypePool(partitionTypeJson) idx }) taskBuilder.setPartitionSpecIdx(specIdx) @@ -393,84 +452,6 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit val partitionData = partitionMethod.invoke(task) if (partitionData != null) { - // Get the partition type/schema from the spec - val partitionTypeMethod = spec.getClass.getMethod("partitionType") - val partitionType = partitionTypeMethod.invoke(spec) - - // Check if partition type has any fields before serializing - val fieldsMethod = partitionType.getClass.getMethod("fields") - val fields = fieldsMethod - .invoke(partitionType) - .asInstanceOf[java.util.List[_]] - - // Helper to get field type string (shared by both type and data serialization) - def getFieldType(field: Any): String = { - val typeMethod = field.getClass.getMethod("type") - typeMethod.invoke(field).toString - } - - // Only serialize partition type if there are actual partition fields - if (!fields.isEmpty) { - try { - // Manually build StructType JSON to match iceberg-rust expectations. - // Using Iceberg's SchemaParser.toJson() would include schema-level - // metadata (e.g., "schema-id") that iceberg-rust's StructType - // deserializer rejects. We need pure StructType format: - // {"type":"struct","fields":[...]} - - // Filter out fields with unknown types (dropped partition fields). - // Unknown type fields represent partition columns that have been dropped - // from the schema. Per the Iceberg spec, unknown type fields are not - // stored in data files and iceberg-rust doesn't support deserializing - // them. Since these columns are dropped, we don't need to expose their - // partition values when reading. - val fieldsJson = fields.asScala.flatMap { field => - val fieldTypeStr = getFieldType(field) - - // Skip fields with unknown type (dropped partition columns) - if (fieldTypeStr == IcebergReflection.TypeNames.UNKNOWN) { - None - } else { - val fieldIdMethod = field.getClass.getMethod("fieldId") - val fieldId = fieldIdMethod.invoke(field).asInstanceOf[Int] - - val nameMethod = field.getClass.getMethod("name") - val fieldName = nameMethod.invoke(field).asInstanceOf[String] - - val isOptionalMethod = field.getClass.getMethod("isOptional") - val isOptional = - isOptionalMethod.invoke(field).asInstanceOf[Boolean] - val required = !isOptional - - Some( - ("id" -> fieldId) ~ - ("name" -> fieldName) ~ - ("required" -> required) ~ - ("type" -> fieldTypeStr)) - } - }.toList - - // Only serialize if we have non-unknown fields - if (fieldsJson.nonEmpty) { - val partitionTypeJson = compact( - render( - ("type" -> "struct") ~ - ("fields" -> fieldsJson))) - - val typeIdx = partitionTypeToPoolIndex.getOrElseUpdate( - partitionTypeJson, { - val idx = partitionTypeToPoolIndex.size - commonBuilder.addPartitionTypePool(partitionTypeJson) - idx - }) - taskBuilder.setPartitionTypeIdx(typeIdx) - } - } catch { - case e: Exception => - logWarning(s"Failed to serialize partition type to JSON: ${e.getMessage}") - } - } - // Serialize partition data to protobuf for native execution. // The native execution engine uses partition_data protobuf messages to // build a constants_map, which provides partition values to identity- @@ -511,13 +492,20 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit // empty partition data out of range. Instead send an empty-fields spec that keeps the // real spec id (so _spec_id stays correct) plus empty data, so native fills every // unified _partition field with null -- matching Spark and the pre-tightening behaviour. + // Pair it with an empty partition-type-pool entry too, keeping the two pools aligned. if (partitionValues.isEmpty) { val specId = spec.getClass.getMethod("specId").invoke(spec).asInstanceOf[Int] - val emptySpecJson = s"""{"spec-id":$specId,"fields":[]}""" + val emptySpecJson = + compact( + render(("spec-id" -> specId) ~ ("fields" -> List.empty[org.json4s.JObject]))) val emptySpecIdx = partitionSpecToPoolIndex.getOrElseUpdate( emptySpecJson, { val idx = partitionSpecToPoolIndex.size commonBuilder.addPartitionSpecPool(emptySpecJson) + val emptyTypeJson = compact( + render(("type" -> "struct") ~ + ("fields" -> List.empty[org.json4s.JObject]))) + commonBuilder.addPartitionTypePool(emptyTypeJson) idx }) // Override the real spec registered above. @@ -885,7 +873,6 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit // Deduplication structures - map unique values to pool indices val schemaToPoolIndex = mutable.HashMap[AnyRef, Int]() - val partitionTypeToPoolIndex = mutable.HashMap[String, Int]() val partitionSpecToPoolIndex = mutable.HashMap[String, Int]() val nameMappingToPoolIndex = mutable.HashMap[String, Int]() val projectFieldIdsToPoolIndex = mutable.HashMap[Seq[Int], Int]() @@ -1140,7 +1127,6 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit fileScanTaskClass, taskBuilder, commonBuilder, - partitionTypeToPoolIndex, partitionSpecToPoolIndex, partitionDataToPoolIndex) @@ -1182,7 +1168,6 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit // Log deduplication summary val allPoolSizes = Seq( schemaToPoolIndex.size, - partitionTypeToPoolIndex.size, partitionSpecToPoolIndex.size, nameMappingToPoolIndex.size, projectFieldIdsToPoolIndex.size, diff --git a/spark/src/test/resources/sql-tests/iceberg/metadata_column_partition.sql b/spark/src/test/resources/sql-tests/iceberg/metadata_column_partition.sql index 7f226a036e..9ceda6bb8e 100644 --- a/spark/src/test/resources/sql-tests/iceberg/metadata_column_partition.sql +++ b/spark/src/test/resources/sql-tests/iceberg/metadata_column_partition.sql @@ -312,8 +312,12 @@ SELECT id, _pos FROM test_cat.db.mor_delete ORDER BY id statement DROP TABLE test_cat.db.mor_delete --- NOTE: Merge-on-read UPDATE is not tested here because UPDATE TABLE is not --- supported in Spark 4.0. Covered in CometIcebergNativeSuite instead. +-- NOTE: Merge-on-read UPDATE is not tested here because UPDATE TABLE is not supported in Spark +-- 4.0, and is not covered anywhere else either: it requires Iceberg's SQL extensions +-- (IcebergSparkSessionExtensions), which are registered at SparkSession construction time and +-- are not wired into CometTestBase's shared test session (same blocker as ALTER TABLE ADD +-- PARTITION FIELD elsewhere in this test suite). Adding coverage needs that session-level +-- infra, not just a new test case. -- ============================================================= -- Temporal transforms: years() on DATE column diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala index 9a20a70a13..c816255924 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala @@ -39,7 +39,7 @@ 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.iceberg.RESTCatalogHelper +import org.apache.comet.iceberg.{IcebergReflection, RESTCatalogHelper} import org.apache.comet.serde.OperatorOuterClass import org.apache.comet.testing.{FuzzDataGenerator, SchemaGenOptions} @@ -4806,6 +4806,41 @@ class CometIcebergNativeSuite } } + test("metadata columns - still-unsupported column falls back to Spark") { + 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.unsupported_meta_col" + try { + spark.sql(s""" + CREATE TABLE $table (id INT, value DOUBLE) USING iceberg + TBLPROPERTIES ('format-version' = '2') + """) + + spark.sql(s"INSERT INTO $table VALUES (1, 10.5), (2, 20.3)") + + // _deleted is a real, selectable Iceberg metadata column (SparkTable.metadataColumns()) + // but is not in CometIcebergNativeScan.MetadataFieldIds, so this must fall back rather + // than silently drop the column or crash. Guards against a future change to + // MetadataFieldIds or the unsupportedMetadataCols filter regressing this path. + checkIcebergNativeScanFallback( + s"SELECT id, _deleted FROM $table ORDER BY id", + "_deleted is not in CometIcebergNativeScan.MetadataFieldIds") + } finally { + spark.sql(s"DROP TABLE IF EXISTS $table PURGE") + } + } + } + } + test("metadata columns - _pos returns row position within each file") { assume(icebergAvailable, "Iceberg not available in classpath") @@ -5365,4 +5400,146 @@ class CometIcebergNativeSuite } } } + + test("partition evolution - conflicting V1 spec field detected before native scan") { + // Iceberg Java's own write path (e.g. updateSpec()) can never produce two specs that bind + // the same partition field id to different source columns: TableMetadata.Builder always + // reassigns fresh, non-conflicting field ids when a spec is added. Only V1 tables written by + // a non-Java writer (or, as here, a hand-edited metadata.json) can have this defect -- + // exactly the case IcebergReflection.validateUnifiedPartitionType/CometScanRule's + // unifiedPartitionTypeSupported guards against. Table *loading* does not validate this + // (TableMetadataParser binds all non-default specs via UnboundPartitionSpec#bindUnchecked), + // so the conflict only surfaces when something asks for the unified partition type. + // + // That "something" turns out to always be Spark itself, not just Comet: resolving + // _partition's output type for any query that touches metadata columns on this table goes + // through SparkTable.metadataColumns(), which eagerly calls the same + // Partitioning.partitionType(table) and throws during analysis, before CometScanRule (which + // runs during physical planning) ever sees the query. So there's no SQL query that + // reaches CometScanRule's fallback for this case -- plain Spark can't analyze it either. This + // test instead exercises IcebergReflection.validateUnifiedPartitionType directly, which is + // the reflection call unifiedPartitionTypeSupported relies on to fall back safely in the + // (currently unreached, but not guaranteed to stay that way across Iceberg versions) case + // where analysis succeeds but the native scan's own merge would otherwise fail. + 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.conflicting_spec" + try { + spark.sql(s""" + CREATE TABLE $table (id INT, region STRING, ts TIMESTAMP) + USING iceberg PARTITIONED BY (region) + TBLPROPERTIES ('format-version' = '1') + """) + spark.sql(s"INSERT INTO $table VALUES (1, 'US', TIMESTAMP '2024-01-01 00:00:00')") + + import org.apache.iceberg.catalog.TableIdentifier + import org.apache.iceberg.spark.SparkCatalog + + val sparkCatalog = spark.sessionState.catalogManager + .catalog("test_cat") + .asInstanceOf[SparkCatalog] + val iceTable = sparkCatalog + .icebergCatalog() + .loadTable(TableIdentifier.of("db", "conflicting_spec")) + + // Derive the metadata dir from the table's own reported location rather than + // reconstructing the warehouse layout by hand, since location() may or may not carry a + // "file:" URI scheme depending on the FileIO in use. + 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") + // Discover the current metadata version by listing files rather than trusting + // version-hint.text: HadoopCatalog (unlike plain HadoopTables) does not always write + // it, and Iceberg's own HadoopTableOperations falls back to a directory scan for the + // same reason. + 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] + val specs = root + .get("partition-specs") + .asInstanceOf[com.fasterxml.jackson.databind.node.ArrayNode] + val spec0 = specs.get(0) + val regionFieldId = spec0.get("fields").get(0).get("field-id").asInt() + val idFieldId = root + .get("schemas") + .get(0) + .get("fields") + .elements() + .asScala + .find(_.get("name").asText() == "id") + .get + .get("id") + .asInt() + + // New spec (id 1) reuses region's field id for a field bound to a *different* source + // column ("id" instead of "region") -- exactly what Partitioning.partitionType()/ + // compute_unified_partition_type's equivalentIgnoringNames rejects. + val conflictingSpec = mapper.createObjectNode() + conflictingSpec.put("spec-id", 1) + val conflictingFields = mapper.createArrayNode() + val conflictingField = mapper.createObjectNode() + conflictingField.put("source-id", idFieldId) + conflictingField.put("field-id", regionFieldId) + conflictingField.put("name", "id_as_region") + conflictingField.put("transform", "identity") + conflictingFields.add(conflictingField) + conflictingSpec.set("fields", conflictingFields) + specs.add(conflictingSpec) + + // Round-trip through Iceberg's own parser before writing, so a malformed hand-edit + // fails this test with a clear parse error rather than producing an unloadable table. + val newMetadataJson = mapper.writeValueAsString(root) + org.apache.iceberg.TableMetadataParser.fromJson(newMetadataJson) + + val newVersion = currentVersion + 1 + val newMetadataFile = new File(metadataDir, s"v$newVersion.metadata.json") + java.nio.file.Files.write(newMetadataFile.toPath, newMetadataJson.getBytes(UTF_8)) + // Best-effort: some catalog configurations track the current version via this file; + // others (see the directory-listing fallback above) don't require it. + java.nio.file.Files.write( + new File(metadataDir, "version-hint.text").toPath, + newVersion.toString.getBytes(UTF_8)) + + // Not testable end-to-end via a SQL query: resolving _partition's output type for + // *any* query touching metadata columns on this table goes through Spark's own + // SparkTable.metadataColumns() -> SparkMetadataColumns.partition() -> + // Partitioning.partitionType(table), which throws this exact ValidationException + // during analysis -- before CometScanRule (which runs during physical planning) ever + // sees the query. So this exercises the same reflection call CometScanRule's + // unifiedPartitionTypeSupported guards with, directly, against the conflicting table. + iceTable.refresh() + val reason = IcebergReflection.validateUnifiedPartitionType(iceTable) + assert( + reason.isDefined, + "Expected a validation failure for conflicting partition specs") + assert( + reason.get.contains("Conflicting partition fields"), + s"Expected a conflicting-partition-fields reason, got: ${reason.get}") + } finally { + spark.sql(s"DROP TABLE IF EXISTS $table PURGE") + } + } + } + } } From 53487f4f2a8268fef818079b4d0a836bb7c0558a Mon Sep 17 00:00:00 2001 From: Parth Chandra Date: Thu, 30 Jul 2026 19:53:09 -0700 Subject: [PATCH 3/6] fix test --- native/core/src/execution/planner.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs index 61c8195846..ed1794b1d3 100644 --- a/native/core/src/execution/planner.rs +++ b/native/core/src/execution/planner.rs @@ -4461,6 +4461,7 @@ mod tests { use crate::execution::operators::ExecutionError; use crate::execution::planner::literal_to_array_ref; + use crate::execution::planner::parse_file_scan_tasks_from_common; use crate::parquet::parquet_support::SparkParquetOptions; use crate::parquet::schema_adapter::SparkPhysicalExprAdapterFactory; use datafusion_comet_proto::spark_expression::expr::ExprStruct; From b2aaae7a9c20bbe677587e704ac506506e898fb8 Mon Sep 17 00:00:00 2001 From: Parth Chandra Date: Fri, 31 Jul 2026 06:55:03 -0700 Subject: [PATCH 4/6] fix --- dev/diffs/iceberg/1.10.0.diff | 14 +++++ dev/diffs/iceberg/1.8.1.diff | 14 +++++ dev/diffs/iceberg/1.9.1.diff | 14 +++++ native/core/src/execution/planner.rs | 80 ++++++++++++++++++++++++---- 4 files changed, 112 insertions(+), 10 deletions(-) diff --git a/dev/diffs/iceberg/1.10.0.diff b/dev/diffs/iceberg/1.10.0.diff index b7c5ce8a84..775f190e99 100644 --- a/dev/diffs/iceberg/1.10.0.diff +++ b/dev/diffs/iceberg/1.10.0.diff @@ -1519,6 +1519,20 @@ index 4c1a509591..b22f3f9ad9 100644 .enableHiveSupport() .getOrCreate(); +diff --git a/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/SparkRowLevelOperationsTestBase.java b/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/SparkRowLevelOperationsTestBase.java +index 893f9931cf..17e5fef217 100644 +--- a/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/SparkRowLevelOperationsTestBase.java ++++ b/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/SparkRowLevelOperationsTestBase.java +@@ -460,7 +460,8 @@ public abstract class SparkRowLevelOperationsTestBase extends ExtensionsTestBase + + protected void assertAllBatchScansVectorized(SparkPlan plan) { + List batchScans = SparkPlanUtil.collectBatchScans(plan); +- assertThat(batchScans).hasSizeGreaterThan(0).allMatch(SparkPlan::supportsColumnar); ++ // When Comet is enabled, its native scan replaces BatchScanExec nodes entirely ++ assertThat(batchScans).allMatch(SparkPlan::supportsColumnar); + } + + protected void createTableWithDeleteGranularity( diff --git a/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCallStatementParser.java b/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCallStatementParser.java index ecf9e6f8a5..23dbcf7af2 100644 --- a/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCallStatementParser.java diff --git a/dev/diffs/iceberg/1.8.1.diff b/dev/diffs/iceberg/1.8.1.diff index 7c951c7f5b..7456bb745e 100644 --- a/dev/diffs/iceberg/1.8.1.diff +++ b/dev/diffs/iceberg/1.8.1.diff @@ -78,6 +78,20 @@ index 4f137f5b8d..145ef2d00c 100644 .getOrCreate(); SparkTestBase.catalog = +diff --git a/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/SparkRowLevelOperationsTestBase.java b/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/SparkRowLevelOperationsTestBase.java +index 5fa76f9da6..9184aaba83 100644 +--- a/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/SparkRowLevelOperationsTestBase.java ++++ b/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/SparkRowLevelOperationsTestBase.java +@@ -452,7 +452,8 @@ public abstract class SparkRowLevelOperationsTestBase extends SparkExtensionsTes + + protected void assertAllBatchScansVectorized(SparkPlan plan) { + List batchScans = SparkPlanUtil.collectBatchScans(plan); +- assertThat(batchScans).hasSizeGreaterThan(0).allMatch(SparkPlan::supportsColumnar); ++ // When Comet is enabled, its native scan replaces BatchScanExec nodes entirely ++ assertThat(batchScans).allMatch(SparkPlan::supportsColumnar); + } + + protected void createTableWithDeleteGranularity( diff --git a/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCallStatementParser.java b/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCallStatementParser.java index 55a413063e..a1895c2faf 100644 --- a/spark/v3.4/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCallStatementParser.java diff --git a/dev/diffs/iceberg/1.9.1.diff b/dev/diffs/iceberg/1.9.1.diff index e086b43a30..32c4e693fc 100644 --- a/dev/diffs/iceberg/1.9.1.diff +++ b/dev/diffs/iceberg/1.9.1.diff @@ -1631,6 +1631,20 @@ index 578845e3da..3cc1598035 100644 .enableHiveSupport() .getOrCreate(); +diff --git a/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/SparkRowLevelOperationsTestBase.java b/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/SparkRowLevelOperationsTestBase.java +index b5d6415763..0763a723d3 100644 +--- a/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/SparkRowLevelOperationsTestBase.java ++++ b/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/SparkRowLevelOperationsTestBase.java +@@ -448,7 +448,8 @@ public abstract class SparkRowLevelOperationsTestBase extends ExtensionsTestBase + + protected void assertAllBatchScansVectorized(SparkPlan plan) { + List batchScans = SparkPlanUtil.collectBatchScans(plan); +- assertThat(batchScans).hasSizeGreaterThan(0).allMatch(SparkPlan::supportsColumnar); ++ // When Comet is enabled, its native scan replaces BatchScanExec nodes entirely ++ assertThat(batchScans).allMatch(SparkPlan::supportsColumnar); + } + + protected void createTableWithDeleteGranularity( diff --git a/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCallStatementParser.java b/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCallStatementParser.java index ecf9e6f8a5..23dbcf7af2 100644 --- a/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCallStatementParser.java diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs index ed1794b1d3..5accd31609 100644 --- a/native/core/src/execution/planner.rs +++ b/native/core/src/execution/planner.rs @@ -3869,17 +3869,19 @@ fn parse_file_scan_tasks_from_common( "Failed to deserialize partition type JSON from pool: {e}" )) })?; - let spec_id = partition_spec_cache + // 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. + let spec_id = proto_common + .partition_spec_pool .get(idx) - .and_then(|opt| opt.as_ref()) - .map(|spec| spec.spec_id()) - .ok_or_else(|| { - ExecutionError::GeneralError(format!( - "No partition spec at pool index {idx} matching partition_type_pool \ - entry; partition_type_pool must be index-aligned with \ - partition_spec_pool" - )) - })?; + .and_then(|json| serde_json::from_str::(json).ok()) + .and_then(|v| v.get("spec-id").and_then(serde_json::Value::as_i64)) + .map(|id| id as i32) + .unwrap_or(idx as i32); Ok((spec_id, struct_type)) }) .collect::, ExecutionError>>()?; @@ -5657,4 +5659,62 @@ mod tests { assert_eq!(fields[0].name, "region_new"); assert_eq!(fields[1].name, "category"); } + + #[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. + let schema_json = serde_json::to_string( + &iceberg::spec::Schema::builder() + .with_schema_id(0) + .with_fields(vec![iceberg::spec::NestedField::required( + 1, + "id", + iceberg::spec::Type::Primitive(iceberg::spec::PrimitiveType::Int), + ) + .into()]) + .build() + .expect("schema"), + ) + .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). + 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":[]}"#; + + let proto_common = spark_operator::IcebergScanCommon { + schema_pool: vec![schema_json], + partition_type_pool: vec![empty_type_json.to_string()], + partition_spec_pool: vec![unparseable_spec_json.to_string()], + project_field_ids_pool: vec![spark_operator::ProjectFieldIdList { + field_ids: vec![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), + project_field_ids_idx: 0, + ..Default::default() + }; + + let tasks = parse_file_scan_tasks_from_common(&proto_common, &[proto_task]) + .expect("parse tasks must not error on an unparseable spec"); + assert_eq!(tasks.len(), 1); + assert!( + tasks[0] + .unified_partition_type + .as_ref() + .map(|t| t.fields().is_empty()) + .unwrap_or(true), + "unified partition type should have no fields for an all-unknown-transform spec" + ); + } } From d37951e72b9e7e677ebd7d354d60be5a533ab012 Mon Sep 17 00:00:00 2001 From: Parth Chandra Date: Fri, 31 Jul 2026 09:44:50 -0700 Subject: [PATCH 5/6] address review: skip real-spec serialization for value-less tasks serializePartitionData computed the file's real partition spec (toJson reflection + pool intern) unconditionally, then overwrote it with the empty-spec index for value-less tasks (partition-evolution files with a dropped field). Move the real-spec serialization into a helper that only the value-carrying path calls, so value-less tasks compute just the empty spec. Spec and partition-type pools stay index-aligned: each path interns both entries together. --- .../operator/CometIcebergNativeScan.scala | 89 +++++++++++-------- 1 file changed, 54 insertions(+), 35 deletions(-) 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 6c4f45ce62..93b4f0de4b 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 @@ -411,40 +411,49 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit } }.toList - // Deduplicate partition spec, pairing each new spec-pool entry with a partition-type-pool - // entry at the same index. Native correlates the two by index to recover, per spec, the - // spec_id needed to merge partition types across historical specs in the same order - // Iceberg Java's Partitioning.partitionType()/iceberg-rust's compute_unified_partition_type - // do (newest spec_id first); see parse_file_scan_tasks_from_common in planner.rs. - try { - val partitionSpecParserClass = - IcebergReflection.loadClass(IcebergReflection.ClassNames.PARTITION_SPEC_PARSER) - val toJsonMethod = partitionSpecParserClass.getMethod( - "toJson", - IcebergReflection.loadClass(IcebergReflection.ClassNames.PARTITION_SPEC)) - val partitionSpecJson = toJsonMethod - .invoke(null, spec) - .asInstanceOf[String] - - val specIdx = partitionSpecToPoolIndex.getOrElseUpdate( - partitionSpecJson, { - val idx = partitionSpecToPoolIndex.size - commonBuilder.addPartitionSpecPool(partitionSpecJson) - // Manually build StructType JSON to match iceberg-rust expectations. - // Using Iceberg's SchemaParser.toJson() would include schema-level - // metadata (e.g., "schema-id") that iceberg-rust's StructType - // deserializer rejects. We need pure StructType format: - // {"type":"struct","fields":[...]} - val partitionTypeJson = compact( - render(("type" -> "struct") ~ - ("fields" -> fieldsJson))) - commonBuilder.addPartitionTypePool(partitionTypeJson) - idx - }) - taskBuilder.setPartitionSpecIdx(specIdx) - } catch { - case e: Exception => - logWarning(s"Failed to serialize partition spec to JSON: ${e.getMessage}") + // Serializes the file's real partition spec plus its paired partition-type-pool entry, + // then points the task at that pool index. Only invoked for tasks that actually carry + // partition values: a value-less task (partition evolution) takes the empty-spec path + // below instead, so this avoids the toJson reflection call and pool intern that would + // otherwise be computed and then immediately overwritten. + // + // The spec and type pools are index-aligned: native correlates the two by index to + // recover, per spec, the spec_id needed to merge partition types across historical specs + // in the same order Iceberg Java's Partitioning.partitionType()/iceberg-rust's + // compute_unified_partition_type do (newest spec_id first); see + // parse_file_scan_tasks_from_common in planner.rs. Every path that adds a spec-pool entry + // adds its type-pool entry in the same block to keep them aligned. + def serializeRealSpec(): Unit = { + try { + val partitionSpecParserClass = + IcebergReflection.loadClass(IcebergReflection.ClassNames.PARTITION_SPEC_PARSER) + val toJsonMethod = partitionSpecParserClass.getMethod( + "toJson", + IcebergReflection.loadClass(IcebergReflection.ClassNames.PARTITION_SPEC)) + val partitionSpecJson = toJsonMethod + .invoke(null, spec) + .asInstanceOf[String] + + val specIdx = partitionSpecToPoolIndex.getOrElseUpdate( + partitionSpecJson, { + val idx = partitionSpecToPoolIndex.size + commonBuilder.addPartitionSpecPool(partitionSpecJson) + // Manually build StructType JSON to match iceberg-rust expectations. + // Using Iceberg's SchemaParser.toJson() would include schema-level + // metadata (e.g., "schema-id") that iceberg-rust's StructType + // deserializer rejects. We need pure StructType format: + // {"type":"struct","fields":[...]} + val partitionTypeJson = compact( + render(("type" -> "struct") ~ + ("fields" -> fieldsJson))) + commonBuilder.addPartitionTypePool(partitionTypeJson) + idx + }) + taskBuilder.setPartitionSpecIdx(specIdx) + } catch { + case e: Exception => + logWarning(s"Failed to serialize partition spec to JSON: ${e.getMessage}") + } } // Get partition data from the task (via file().partition()) @@ -493,6 +502,9 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit // real spec id (so _spec_id stays correct) plus empty data, so native fills every // unified _partition field with null -- matching Spark and the pre-tightening behaviour. // Pair it with an empty partition-type-pool entry too, keeping the two pools aligned. + // + // Only the value-carrying path serializes the real spec (serializeRealSpec), so the + // value-less path never does the reflection/intern work just to overwrite it. if (partitionValues.isEmpty) { val specId = spec.getClass.getMethod("specId").invoke(spec).asInstanceOf[Int] val emptySpecJson = @@ -508,8 +520,9 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit commonBuilder.addPartitionTypePool(emptyTypeJson) idx }) - // Override the real spec registered above. taskBuilder.setPartitionSpecIdx(emptySpecIdx) + } else { + serializeRealSpec() } // Always send partition data (empty when there are no values) so native never sees a @@ -531,6 +544,12 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit idx }) taskBuilder.setPartitionDataIdx(partitionDataIdx) + } else { + // Defensive: ContentScanTask.partition() returns an empty struct (never null) for + // unpartitioned tables in practice. If it is ever null we cannot compute values, so + // just register the file's real spec (preserving the pre-refactor behaviour) to keep + // _spec_id correct. + serializeRealSpec() } } } catch { From 967937176359d8c62bd93eb412bcf28c5db5ba49 Mon Sep 17 00:00:00 2001 From: Parth Chandra Date: Fri, 31 Jul 2026 09:51:37 -0700 Subject: [PATCH 6/6] address review: describe native reader behaviour, not the storage backend The FileIO-compatibility comments named object_store as the backend, but the Iceberg scan path reads through iceberg-rust's own storage layer, not object_store. Since iceberg-rust's storage backend may change on a dependency bump, describe the behaviour (bypasses Iceberg Java FileIO, reads through iceberg-rust's storage layer) rather than naming a backend. Non-Iceberg V1 Parquet/CSV comments that genuinely use object_store are left unchanged. --- .../org/apache/comet/iceberg/IcebergReflection.scala | 5 +++-- .../main/scala/org/apache/comet/rules/CometScanRule.scala | 8 ++++---- 2 files changed, 7 insertions(+), 6 deletions(-) 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 7e8ba9df88..f867d4bc9b 100644 --- a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala +++ b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala @@ -68,7 +68,7 @@ object IcebergReflection extends Logging { def isIcebergScanClass(name: String): Boolean = ICEBERG_SCAN_CLASSES.contains(name) - // Iceberg FileIO implementations whose underlying storage object_store can reach. + // Iceberg FileIO implementations whose backing storage Comet's native reader can reach. // Custom/test FileIO classes (e.g. CustomFileIO in TestSparkExecutorCache) are not compatible // because Comet's native reader bypasses Java FileIO entirely. val COMPATIBLE_FILE_IO_CLASSES: Set[String] = Set( @@ -84,7 +84,8 @@ object IcebergReflection extends Logging { // EncryptingFileIO but a nested variant chosen from the wrapped delegate's capabilities // (e.g. EncryptingFileIO$WithSupportsPrefixOperations when the delegate is HadoopFileIO), so // an exact class-name match misses it. Comet forwards each file's key_metadata to iceberg-rust - // and reads the ciphertext via object_store, so any EncryptingFileIO variant is compatible. + // and reads the ciphertext through iceberg-rust's own storage layer, so any EncryptingFileIO + // variant is compatible. private val ENCRYPTING_FILE_IO_PREFIX = "org.apache.iceberg.encryption.EncryptingFileIO" /** 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 41bc996052..ca917335da 100644 --- a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala +++ b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala @@ -473,15 +473,15 @@ case class CometScanRule(session: SparkSession) // Now perform all validation using the pre-extracted metadata // Check if table uses a FileIO implementation compatible with iceberg-rust. - // Comet's native reader uses object_store (Rust) for I/O, bypassing Iceberg Java's - // FileIO entirely. Only allow known-compatible implementations whose underlying - // storage object_store can reach via standard URL schemes. + // Comet's native reader bypasses Iceberg Java's FileIO entirely and reads through + // iceberg-rust's own storage layer instead. Only allow known-compatible implementations + // whose backing storage that layer can reach via standard URL schemes. val fileIOCompatible = IcebergReflection.getFileIO(metadata.table) match { case Some(fileIO) if IcebergReflection.isCompatibleFileIO(fileIO) => true case Some(fileIO) => fallbackReasons += s"FileIO ${fileIO.getClass.getName} is not supported by " + - "Comet's native reader (object_store bypasses Iceberg Java FileIO)" + "Comet's native reader (bypasses Iceberg Java FileIO)" false case None => fallbackReasons += "Could not check FileIO compatibility"