diff --git a/dev/diffs/iceberg/1.11.0.diff b/dev/diffs/iceberg/1.11.0.diff index a404ded20a..ca56ab9b93 100644 --- a/dev/diffs/iceberg/1.11.0.diff +++ b/dev/diffs/iceberg/1.11.0.diff @@ -251,6 +251,31 @@ index 507d7b313b..3b73dcc014 100644 .config("spark.ui.enabled", "false") .config(DISABLE_UI) .enableHiveSupport() +diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/data/AvroDataTestBase.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/data/AvroDataTestBase.java +index 8ce60f6275..75112b4354 100644 +--- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/data/AvroDataTestBase.java ++++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/data/AvroDataTestBase.java +@@ -48,7 +48,6 @@ import org.apache.iceberg.types.Types.LongType; + import org.apache.iceberg.types.Types.MapType; + import org.apache.iceberg.types.Types.StructType; + import org.apache.iceberg.util.DateTimeUtil; +-import org.assertj.core.api.Condition; + import org.junit.jupiter.api.Test; + import org.junit.jupiter.api.io.TempDir; + import org.junit.jupiter.params.ParameterizedTest; +@@ -385,12 +384,6 @@ public abstract class AvroDataTestBase { + .build()); + + assertThatThrownBy(() -> writeAndValidate(writeSchema, expectedSchema)) +- .has( +- new Condition<>( +- t -> +- IllegalArgumentException.class.isInstance(t) +- || IllegalArgumentException.class.isInstance(t.getCause()), +- "Expecting a throwable or cause that is an instance of IllegalArgumentException")) + .hasMessageContaining("Missing required field: missing_str"); + } + diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/data/vectorized/parquet/TestParquetDictionaryEncodedVectorizedReads.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/data/vectorized/parquet/TestParquetDictionaryEncodedVectorizedReads.java index b61ecfa2f4..d696e85139 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/data/vectorized/parquet/TestParquetDictionaryEncodedVectorizedReads.java diff --git a/native/Cargo.lock b/native/Cargo.lock index c54812c818..99ee4a5d77 100644 --- a/native/Cargo.lock +++ b/native/Cargo.lock @@ -3602,7 +3602,7 @@ dependencies = [ [[package]] name = "iceberg" version = "0.10.0" -source = "git+https://github.com/apache/iceberg-rust?rev=ff678ca#ff678ca6d1123d9d82902b34379ccba2caf172b6" +source = "git+https://github.com/mbutrovich/iceberg-rust?rev=108f8d8682822738db753bb92fca1ec43cca8bf2#108f8d8682822738db753bb92fca1ec43cca8bf2" dependencies = [ "aes-gcm", "anyhow", @@ -3623,6 +3623,7 @@ dependencies = [ "bimap", "bytes", "chrono", + "crc32fast", "derive_builder", "expect-test", "fastnum", @@ -3658,7 +3659,7 @@ dependencies = [ [[package]] name = "iceberg-storage-opendal" version = "0.10.0" -source = "git+https://github.com/apache/iceberg-rust?rev=ff678ca#ff678ca6d1123d9d82902b34379ccba2caf172b6" +source = "git+https://github.com/mbutrovich/iceberg-rust?rev=108f8d8682822738db753bb92fca1ec43cca8bf2#108f8d8682822738db753bb92fca1ec43cca8bf2" dependencies = [ "anyhow", "async-trait", diff --git a/native/Cargo.toml b/native/Cargo.toml index e2277596aa..acf4e98247 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 = "ff678ca" } -iceberg-storage-opendal = { git = "https://github.com/apache/iceberg-rust", rev = "ff678ca", features = ["opendal-memory", "opendal-fs", "opendal-s3", "opendal-gcs", "opendal-oss", "opendal-azdls"] } +iceberg = { git = "https://github.com/mbutrovich/iceberg-rust", rev = "108f8d8682822738db753bb92fca1ec43cca8bf2" } +iceberg-storage-opendal = { git = "https://github.com/mbutrovich/iceberg-rust", rev = "108f8d8682822738db753bb92fca1ec43cca8bf2", 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 d89a25c879..0ba5fa0c45 100644 --- a/native/core/src/execution/operators/iceberg_scan.rs +++ b/native/core/src/execution/operators/iceberg_scan.rs @@ -670,6 +670,9 @@ mod tests { file_size_in_bytes: 0, partition_spec_id: 0, equality_ids: None, + referenced_data_file: None, + content_offset: None, + content_size_in_bytes: None, } } diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs index 7ed2b3331c..2ead381ad1 100644 --- a/native/core/src/execution/planner.rs +++ b/native/core/src/execution/planner.rs @@ -3699,6 +3699,9 @@ fn parse_file_scan_tasks_from_common( } else { Some(del.equality_ids.clone()) }, + referenced_data_file: del.referenced_data_file.clone(), + content_offset: del.content_offset, + content_size_in_bytes: del.content_size_in_bytes, }) }) .collect::, ExecutionError>>() diff --git a/native/proto/src/proto/operator.proto b/native/proto/src/proto/operator.proto index 2fcfe7f25b..917e2923ef 100644 --- a/native/proto/src/proto/operator.proto +++ b/native/proto/src/proto/operator.proto @@ -277,6 +277,13 @@ message IcebergDeleteFile { // Equality field IDs (empty for positional deletes) repeated int32 equality_ids = 4; + + // Deletion vector coordinates, set only for V3 deletion vectors. referenced_data_file is the + // data file the vector applies to; content_offset and content_size_in_bytes locate the + // deletion-vector-v1 blob within its Puffin file. + optional string referenced_data_file = 5; + optional int64 content_offset = 6; + optional int64 content_size_in_bytes = 7; } message Projection { 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 9f48ddfa9e..6dd78fd745 100644 --- a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala +++ b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala @@ -51,6 +51,7 @@ object IcebergReflection extends Logging { val UNBOUND_PREDICATE = "org.apache.iceberg.expressions.UnboundPredicate" 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" } /** @@ -78,6 +79,7 @@ object IcebergReflection extends Logging { */ object FileFormats { val PARQUET = "PARQUET" + val PUFFIN = "PUFFIN" } /** @@ -735,6 +737,63 @@ object IcebergReflection extends Logging { unsupportedTypes.toList } + + /** + * True if the table has table-level encryption configured. Encrypted Parquet footers cannot be + * read by Comet's native reader, so such tables must fall back to Spark. + */ + def hasTableEncryption(table: Any): Boolean = + getTableProperties(table).exists(_.containsKey("encryption.key-id")) + + /** + * Returns the names of schema columns (including nested struct/list/map fields) that declare a + * V3 initial-default value. iceberg-rust does not synthesize default values for columns absent + * from a data file, so reads projecting such columns must fall back. Throws on reflection + * failure so the caller can fall back rather than risk a native crash. + */ + def columnsWithInitialDefault(schema: Any): List[String] = { + import scala.jdk.CollectionConverters._ + val columns = + schema.getClass.getMethod("columns").invoke(schema).asInstanceOf[java.util.List[_]] + columns.asScala.flatMap(walkFieldForDefault).toList + } + + private def walkFieldForDefault(field: Any): List[String] = { + import scala.jdk.CollectionConverters._ + val name = field.getClass.getMethod("name").invoke(field).asInstanceOf[String] + val here = + if (field.getClass.getMethod("initialDefault").invoke(field) != null) List(name) else Nil + val fieldType = field.getClass.getMethod("type").invoke(field) + val nested = + if (fieldType.getClass.getMethod("isNestedType").invoke(fieldType).asInstanceOf[Boolean]) { + val nestedType = fieldType.getClass.getMethod("asNestedType").invoke(fieldType) + val fields = + nestedType.getClass + .getMethod("fields") + .invoke(nestedType) + .asInstanceOf[java.util.List[_]] + fields.asScala.flatMap(walkFieldForDefault).toList + } else { + Nil + } + here ++ nested + } + + /** + * Converts an Iceberg `Schema` to the Spark `StructType` it reads as, via + * `SparkSchemaUtil.convert`. Comet serializes the whole table/scan schema to native (not just + * projected columns), so callers use this to run the schema through Comet's existing type + * allow-list and fall back if any column is a type the native reader does not support (e.g. + * variant). Throws on reflection failure so the caller can fall back. + */ + def toSparkSchema(schema: Any): org.apache.spark.sql.types.StructType = { + val sparkSchemaUtil = loadClass(ClassNames.SPARK_SCHEMA_UTIL) + val schemaClass = loadClass(ClassNames.SCHEMA) + val convert = sparkSchemaUtil.getMethod("convert", schemaClass) + convert + .invoke(null, schema.asInstanceOf[AnyRef]) + .asInstanceOf[org.apache.spark.sql.types.StructType] + } } /** 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 59331129d5..19cffa3b62 100644 --- a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala +++ b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala @@ -462,12 +462,16 @@ case class CometScanRule(session: SparkSession) // Check Iceberg table format version + // V3 adds column types iceberg-rust cannot read (variant, geometry, geography, unknown). + // We do not enumerate them here: the readSchema allow-list in DataTypeSupport already + // falls back on any type Comet does not support, which is more future-proof than a deny + // list. This gate only bounds the format version. val formatVersionSupported = IcebergReflection.getFormatVersion(metadata.table) match { case Some(formatVersion) => - if (formatVersion > 2) { + if (formatVersion > 3) { fallbackReasons += "Iceberg table format version " + s"$formatVersion is not supported. " + - "Comet only supports Iceberg table format V1 and V2" + "Comet supports Iceberg table format V1, V2, and V3" false } else { true @@ -477,6 +481,48 @@ case class CometScanRule(session: SparkSession) false } + // Encrypted Parquet footers cannot be read by the native reader. + val encryptionSupported = !IcebergReflection.hasTableEncryption(metadata.table) + if (!encryptionSupported) { + fallbackReasons += "Iceberg table encryption is not supported by Comet's native reader" + } + + // iceberg-rust does not synthesize V3 initial-default values for columns absent from a + // data file, so a scan projecting such a column must fall back. Reflection failure also + // falls back rather than risk a native crash. + val defaultValuesSupported = + try { + val defaulted = IcebergReflection.columnsWithInitialDefault(metadata.scanSchema) + if (defaulted.nonEmpty) { + fallbackReasons += "Iceberg column(s) with V3 default values are not yet supported " + + s"by Comet's native reader: ${defaulted.mkString(", ")}" + false + } else { + true + } + } catch { + case e: Exception => + fallbackReasons += "Iceberg reflection failure: could not verify V3 default " + + s"values: ${e.getMessage}" + false + } + + // Comet serializes the whole table/scan schema to native, not just projected columns, so a + // type the native reader does not support (e.g. variant) breaks the scan even when that + // column is not projected. The readSchema allow-list above only covers projected columns, + // so run the same allow-list over the full schema Comet may serialize. Reflection failure + // also falls back. + val schemaTypesSupported = + try { + val fullSchema = IcebergReflection.toSparkSchema(metadata.tableSchema) + typeChecker.isSchemaSupported(fullSchema, fallbackReasons) + } catch { + case e: Exception => + fallbackReasons += "Iceberg reflection failure: could not verify column " + + s"types: ${e.getMessage}" + false + } + // Single-pass validation of all FileScanTasks val taskValidation = try { @@ -605,15 +651,17 @@ case class CometScanRule(session: SparkSession) if (!taskValidation.deleteFiles.isEmpty) { val historicSchemas = IcebergReflection.getAllSchemas(metadata.table) taskValidation.deleteFiles.asScala.foreach { deleteFile => - // iceberg-rust only reads Parquet delete files. Avro/ORC positional or - // equality deletes must be applied by Spark. + // iceberg-rust reads Parquet delete files (position and equality) and Puffin + // deletion vectors. Avro/ORC deletes must be applied by Spark. IcebergReflection.getFileFormat(deleteFile) match { - case Some(fmt) if fmt.equalsIgnoreCase(IcebergReflection.FileFormats.PARQUET) => + case Some(fmt) + if fmt.equalsIgnoreCase(IcebergReflection.FileFormats.PARQUET) || + fmt.equalsIgnoreCase(IcebergReflection.FileFormats.PUFFIN) => case Some(fmt) => hasUnsupportedDeletes = true fallbackReasons += s"Delete file format '$fmt' is not supported by iceberg-rust. " + - "Only Parquet delete files can be applied natively." + "Only Parquet and Puffin delete files can be applied natively." case None => hasUnsupportedDeletes = true logWarning( @@ -712,10 +760,13 @@ case class CometScanRule(session: SparkSession) } } - if (schemaSupported && fileIOCompatible && formatVersionSupported && + val nativeEligible = schemaSupported && fileIOCompatible && formatVersionSupported && + encryptionSupported && defaultValuesSupported && schemaTypesSupported && taskValidation.allParquet && allSupportedFilesystems && partitionTypesSupported && complexTypePredicatesSupported && transformFunctionsSupported && - deleteFileTypesSupported && dppSubqueriesSupported) { + deleteFileTypesSupported && dppSubqueriesSupported + + if (nativeEligible) { CometBatchScanExec( scanExec.clone().asInstanceOf[BatchScanExec], runtimeFilters = scanExec.runtimeFilters, 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 879762818f..a03e55d7d3 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 @@ -283,6 +283,36 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit case _: Exception => } + // V3 deletion vector coordinates. These DeleteFile accessors exist only on newer Iceberg + // versions and return null for non-deletion-vector deletes, so absence is expected and + // non-fatal. + try { + deleteFileClass.getMethod("referencedDataFile").invoke(deleteFile) match { + case path: String => deleteBuilder.setReferencedDataFile(path) + case _ => + } + } catch { + case _: Exception => + } + + try { + deleteFileClass.getMethod("contentOffset").invoke(deleteFile) match { + case offset: java.lang.Long => deleteBuilder.setContentOffset(offset) + case _ => + } + } catch { + case _: Exception => + } + + try { + deleteFileClass.getMethod("contentSizeInBytes").invoke(deleteFile) match { + case size: java.lang.Long => deleteBuilder.setContentSizeInBytes(size) + case _ => + } + } catch { + case _: Exception => + } + deleteBuilder.build() }.toSeq } catch { diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala index 9656a13c82..a874e53c74 100644 --- a/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala +++ b/spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala @@ -575,6 +575,66 @@ class CometIcebergNativeSuite } } + // V3 tables store positional deletes as deletion vectors (Puffin blobs) instead of position + // delete files. This verifies Comet reads a V3 table's deletion vectors natively and applies + // them, matching Spark. + test("MOR V3 table with DELETION VECTORS - verify deletes are applied") { + assume(icebergAvailable, "Iceberg not available in classpath") + // Format version 3 (and its deletion vectors) requires Iceberg 1.11+. Older Iceberg + // rejects `format-version=3` at table creation. + assume(icebergVersionAtLeast(1, 11), "Iceberg V3 tables require Iceberg 1.11+") + + withTempIcebergDir { warehouseDir => + withSQLConf( + "spark.sql.catalog.test_cat" -> "org.apache.iceberg.spark.SparkCatalog", + "spark.sql.catalog.test_cat.type" -> "hadoop", + "spark.sql.catalog.test_cat.warehouse" -> warehouseDir.getAbsolutePath, + CometConf.COMET_ENABLED.key -> "true", + CometConf.COMET_EXEC_ENABLED.key -> "true", + CometConf.COMET_ICEBERG_NATIVE_ENABLED.key -> "true") { + + spark.sql(""" + CREATE TABLE test_cat.db.dv_delete_test ( + id INT, + name STRING, + value DOUBLE + ) USING iceberg + TBLPROPERTIES ( + 'format-version' = '3', + 'write.delete.mode' = 'merge-on-read', + 'write.merge.mode' = 'merge-on-read' + ) + """) + + spark.sql(""" + INSERT INTO test_cat.db.dv_delete_test + VALUES + (1, 'Alice', 10.5), (2, 'Bob', 20.3), (3, 'Charlie', 30.7), + (4, 'Diana', 15.2), (5, 'Eve', 25.8), (6, 'Frank', 35.0), + (7, 'Grace', 12.1), (8, 'Hank', 22.5) + """) + + spark.sql("DELETE FROM test_cat.db.dv_delete_test WHERE id IN (2, 4, 6)") + + // Confirm the delete produced a deletion vector (a Puffin delete file), not a parquet + // position-delete file or a copy-on-write rewrite, so this test actually exercises the + // deletion-vector read path. + val deleteFormats = spark + .sql("SELECT file_format FROM test_cat.db.dv_delete_test.files WHERE content = 1") + .collect() + .map(_.getString(0)) + .toSet + assert( + deleteFormats.contains("PUFFIN"), + s"expected a deletion vector (PUFFIN delete file) but found: $deleteFormats") + + checkIcebergNativeScan("SELECT * FROM test_cat.db.dv_delete_test ORDER BY id") + + spark.sql("DROP TABLE test_cat.db.dv_delete_test") + } + } + } + test("bytes_scanned includes delete file I/O") { assume(icebergAvailable, "Iceberg not available in classpath")