feat: [DO NOT MERGE] Iceberg table format V3: native deletion vector reads, fall back for other V3 features

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/mbutrovich/iceberg-rust", rev = "5dade7e8f771eb44847ab821589708da269bb97e" }
+iceberg-storage-opendal = { git = "https://github.com/mbutrovich/iceberg-rust", rev = "5dade7e8f771eb44847ab821589708da269bb97e", features = ["opendal-memory", "opendal-fs", "opendal-s3", "opendal-gcs", "opendal-oss", "opendal-azdls"] }
 reqsign-core = "3"
 
 [profile.release]

native/core/src/execution/operators/iceberg_scan.rs

@@ -672,6 +672,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,
             key_metadata: None,
         }
     }

native/core/src/execution/planner.rs

@@ -3712,6 +3712,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,
                 // Plaintext StandardKeyMetadata forwarded verbatim from the JVM; decoded by
                 // iceberg-rust with no KMS unwrap. None for unencrypted delete files.
                 key_metadata: del.key_metadata.clone().map(Vec::into_boxed_slice),

native/proto/src/proto/operator.proto

@@ -350,10 +350,12 @@ message IcebergDeleteFile {
   // Equality field IDs (empty for positional deletes)
   repeated int32 equality_ids = 4;
 
-  // Fields 5-7 are reserved for the deletion-vector coordinates on the dv-read branch
-  // (referenced_data_file / content_offset / content_size_in_bytes) so the two features can merge
-  // in either order without a tag collision.
-  reserved 5, 6, 7;
+  // 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;
 
   // Serialized StandardKeyMetadata for an encrypted delete file, from DeleteFile.keyMetadata().
   // Absent for unencrypted tables. Decoded directly by iceberg-rust; no KMS unwrap needed.

spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala

@@ -79,6 +79,7 @@ object IcebergReflection extends Logging {
    */
   object FileFormats {
     val PARQUET = "PARQUET"
+    val PUFFIN = "PUFFIN"
   }
 
   /**

spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala

@@ -470,8 +470,8 @@ case class CometScanRule(session: SparkSession)
         // V3 adds column types iceberg-rust cannot read (variant, geometry, geography, unknown)
         // and column default values; those are handled by the allow-list and default-value checks
         // below, which fall back per-table. This gate only bounds the format version. Deletion
-        // vectors (a V3 delete feature) are handled by the delete-file gate, which still falls back
-        // for non-Parquet (Puffin) deletes since native deletion-vector reads are not yet wired.
+        // vectors (a V3 delete feature) are read natively; the delete-file gate accepts their
+        // Puffin format.
         val formatVersion = IcebergReflection.getFormatVersion(metadata.table)
         val formatVersionSupported = formatVersion match {
           case Some(v) =>
@@ -678,15 +678,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(

spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala

@@ -313,6 +313,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 =>
+        }
+
         // Encrypted delete files carry a plaintext StandardKeyMetadata blob; forward it verbatim.
         // Unencrypted delete files leave the field unset.
         keyMetadataBytes(keyMetadataMethod, deleteFile).foreach(deleteBuilder.setKeyMetadata)
@@ -853,11 +883,13 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
     val nameMappingToPoolIndex = mutable.HashMap[String, Int]()
     val projectFieldIdsToPoolIndex = mutable.HashMap[Seq[Int], Int]()
     val partitionDataToPoolIndex = mutable.HashMap[String, Int]()
-    // Individual delete files are interned into a flat pool keyed by path (a delete file's path is
-    // its identity); deleteFilesToPoolIndex then dedups the per-task sets as lists of indices into
-    // that pool. One delete file applies to many data files under Iceberg's default partition
-    // delete granularity, so interning avoids re-serializing it once per referencing FileScanTask.
-    val deleteFileToPoolIndex = mutable.HashMap[String, Int]()
+    // Individual delete files are interned into a flat pool keyed by the full delete-file message;
+    // deleteFilesToPoolIndex then dedups the per-task sets as lists of indices into that pool. Path
+    // alone is not an identity: V3 deletion vectors for different data files share one Puffin file
+    // and differ only by content offset, so keying on path would collapse them. One delete file
+    // applies to many data files under Iceberg's default partition delete granularity, so interning
+    // avoids re-serializing it once per referencing FileScanTask.
+    val deleteFileToPoolIndex = mutable.HashMap[OperatorOuterClass.IcebergDeleteFile, Int]()
     val deleteFilesToPoolIndex =
       mutable.HashMap[Seq[Int], Int]()
     val residualToPoolIndex = mutable.HashMap[OperatorOuterClass.IcebergPredicate, Int]()
@@ -1054,7 +1086,7 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
                   // resulting list of pool indices.
                   val deleteFileIndices = deleteFilesList.map { df =>
                     deleteFileToPoolIndex.getOrElseUpdate(
-                      df.getFilePath, {
+                      df, {
                         val idx = deleteFileToPoolIndex.size
                         commonBuilder.addDeleteFilePool(df)
                         idx

spark/src/test/scala/org/apache/comet/CometIcebergNativeSuite.scala

@@ -595,6 +595,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")
+      }
+    }
+  }
+
   // Under Iceberg's default partition delete granularity, one position-delete file applies to every
   // data file in the partition with a compatible sequence number (DeleteFileIndex.forDataFile).
   // Interleaving inserts and deletes staggers data-file sequence numbers, so each FileScanTask sees