fix(scan): read deletion vectors via manifest content_offset/content_size

The deletion vector read path located the DV blob by parsing a Puffin footer (PuffinReader::file_metadata). That only works when the DV is wrapped in a full Puffin container — but the Iceberg spec addresses a DV blob by the manifest entry's content_offset + content_size_in_bytes, and writers such as DuckDB store the deletion-vector-v1 blob standalone (no PFA1 framing). Reading those via the footer fails with 'Bad magic value ... should be PFA1'.

Now FileScanTaskDeleteFile carries content_offset + content_size_in_bytes, and the loader reads exactly those bytes and parses them with the new DeleteVector::from_serialized_bytes (factored out of from_puffin_blob). This works whether the DV is standalone or inside a Puffin container. Validated cross-engine on real data: iceberg-rust scans tables with DuckDB-written DVs and returns the correct post-delete row counts (900K/1M, 9M/10M).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Signed-off-by: Raghvendra Singh <raghav@cashify.in>

Author: raghav-reglobe

crates/iceberg/src/arrow/caching_delete_file_loader.rs

@@ -31,7 +31,6 @@ use crate::delete_vector::DeleteVector;
 use crate::expr::Predicate::AlwaysTrue;
 use crate::expr::{Predicate, Reference};
 use crate::io::FileIO;
-use crate::puffin::{DELETION_VECTOR_V1, PuffinReader};
 use crate::runtime::Runtime;
 use crate::scan::{ArrowRecordBatchStream, FileScanTaskDeleteFile};
 use crate::spec::{
@@ -59,11 +58,13 @@ enum DeleteFileContext {
         file_path: String,
         stream: ArrowRecordBatchStream,
     },
-    /// A V3 deletion vector stored in a Puffin file; the blob is read and parsed
-    /// into a `DeleteVector` in the parse phase.
+    /// A V3 deletion vector; the blob located by `content_offset`/`content_size`
+    /// is read and parsed into a `DeleteVector` in the parse phase.
     DelVec {
         file_path: String,
         referenced_data_file: String,
+        content_offset: u64,
+        content_size: u64,
         file_io: FileIO,
     },
     FreshEqDel {
@@ -258,7 +259,8 @@ impl CachingDeleteFileLoader {
                     }
                     PosDelLoadAction::Load => {
                         if task.file_format == DataFileFormat::Puffin {
-                            // V3 deletion vector — parsed from the Puffin file in the
+                            // V3 deletion vector — the blob located by
+                            // content_offset/content_size is read + parsed in the
                             // parse phase (not a Parquet positional-delete stream).
                             Ok(DeleteFileContext::DelVec {
                                 file_path: task.file_path.clone(),
@@ -271,6 +273,18 @@ impl CachingDeleteFileLoader {
                                             "deletion vector is missing referenced_data_file",
                                         )
                                     })?,
+                                content_offset: task.content_offset.ok_or_else(|| {
+                                    Error::new(
+                                        ErrorKind::DataInvalid,
+                                        "deletion vector is missing content_offset",
+                                    )
+                                })? as u64,
+                                content_size: task.content_size_in_bytes.ok_or_else(|| {
+                                    Error::new(
+                                        ErrorKind::DataInvalid,
+                                        "deletion vector is missing content_size_in_bytes",
+                                    )
+                                })? as u64,
                                 file_io: basic_delete_file_loader.file_io().clone(),
                             })
                         } else {
@@ -338,10 +352,13 @@ impl CachingDeleteFileLoader {
             DeleteFileContext::DelVec {
                 file_path,
                 referenced_data_file,
+                content_offset,
+                content_size,
                 file_io,
             } => {
                 let delete_vector =
-                    Self::parse_deletion_vector_from_puffin(&file_io, &file_path).await?;
+                    Self::read_deletion_vector(&file_io, &file_path, content_offset, content_size)
+                        .await?;
                 let mut results = HashMap::default();
                 results.insert(referenced_data_file, delete_vector);
                 Ok(ParsedDeleteFileContext::DelVecs { file_path, results })
@@ -368,26 +385,34 @@ impl CachingDeleteFileLoader {
         }
     }
 
-    /// Reads a V3 `deletion-vector-v1` blob from a Puffin file into a `DeleteVector`.
-    async fn parse_deletion_vector_from_puffin(
+    /// Reads a V3 deletion-vector-v1 blob located at `content_offset` (length
+    /// `content_size`) within `file_path` and parses it into a `DeleteVector`.
+    ///
+    /// The blob is addressed by the manifest entry's content offset/size rather
+    /// than by parsing a Puffin footer, so this works whether the deletion vector
+    /// is a standalone blob file (e.g. written by DuckDB) or stored inside a
+    /// Puffin container.
+    async fn read_deletion_vector(
         file_io: &FileIO,
         file_path: &str,
+        content_offset: u64,
+        content_size: u64,
     ) -> Result<DeleteVector> {
-        let input = file_io.new_input(file_path)?;
-        let reader = PuffinReader::new(input);
-        let metadata = reader.file_metadata().await?;
-        let blob_metadata = metadata
-            .blobs()
-            .iter()
-            .find(|blob| blob.r#type == DELETION_VECTOR_V1)
+        let bytes = file_io.new_input(file_path)?.read().await?;
+        let start = content_offset as usize;
+        let end = start
+            .checked_add(content_size as usize)
+            .filter(|end| *end <= bytes.len())
             .ok_or_else(|| {
                 Error::new(
                     ErrorKind::DataInvalid,
-                    format!("Puffin file {file_path} contains no deletion-vector-v1 blob"),
+                    format!(
+                        "deletion vector blob range {content_offset}..+{content_size} is out of bounds for {file_path} ({} bytes)",
+                        bytes.len()
+                    ),
                 )
             })?;
-        let blob = reader.blob(blob_metadata).await?;
-        DeleteVector::from_puffin_blob(blob)
+        DeleteVector::from_serialized_bytes(&bytes[start..end])
     }
 
     /// Parses a record batch stream coming from positional delete files
@@ -1099,6 +1124,8 @@ mod tests {
             file_type: DataContentType::PositionDeletes,
             partition_spec_id: 0,
             equality_ids: None,
+            content_offset: dv_data_file.content_offset(),
+            content_size_in_bytes: dv_data_file.content_size_in_bytes(),
             file_format: DataFileFormat::Puffin,
             referenced_data_file: dv_data_file.referenced_data_file(),
         };

crates/iceberg/src/delete_vector.rs

@@ -156,8 +156,17 @@ impl DeleteVector {
                 format!("unsupported puffin blob type: {}", blob.blob_type()),
             ));
         }
+        Self::from_serialized_bytes(blob.data())
+    }
 
-        let data = blob.data();
+    /// Parse a serialized `deletion-vector-v1` blob payload:
+    /// `[u32 BE length][magic 0xD1D33964][serialized RoaringTreemap][u32 BE CRC32]`.
+    ///
+    /// These are exactly the bytes located by a manifest entry's `content_offset`
+    /// and `content_size_in_bytes`, so this works whether the deletion vector is a
+    /// standalone blob file (e.g. written by DuckDB) or stored inside a Puffin
+    /// container (the manifest offset points past the container framing).
+    pub fn from_serialized_bytes(data: &[u8]) -> Result<Self> {
         if data.len() < MIN_SERIALIZED_DELETION_VECTOR_BLOB {
             return Err(Error::new(
                 ErrorKind::DataInvalid,

crates/iceberg/src/scan/task.rs

@@ -163,6 +163,8 @@ impl From<&DeleteFileContext> for FileScanTaskDeleteFile {
             .with_equality_ids(ctx.manifest_entry.data_file.equality_ids.clone())
             .with_file_format(ctx.manifest_entry.data_file().file_format())
             .with_referenced_data_file(ctx.manifest_entry.data_file().referenced_data_file())
+            .with_content_offset(ctx.manifest_entry.data_file().content_offset())
+            .with_content_size_in_bytes(ctx.manifest_entry.data_file().content_size_in_bytes())
             .build()
     }
 }
@@ -196,4 +198,14 @@ pub struct FileScanTaskDeleteFile {
     /// `referenced_data_file`). `None` for positional/equality delete files.
     #[builder(default)]
     pub referenced_data_file: Option<String>,
+
+    /// Offset of the deletion-vector blob within the file (the manifest entry's
+    /// `content_offset`). `None` for positional/equality delete files.
+    #[builder(default)]
+    pub content_offset: Option<i64>,
+
+    /// Size in bytes of the deletion-vector blob (the manifest entry's
+    /// `content_size_in_bytes`). `None` for positional/equality delete files.
+    #[builder(default)]
+    pub content_size_in_bytes: Option<i64>,
 }