fix(scan): decompress non-spec-compliant compressed deletion vectors

The Iceberg spec requires a deletion-vector-v1 blob to be stored uncompressed ('Omit compression-codec; deletion-vector-v1 is not compressed'), so reading the bytes at content_offset/content_size and parsing them directly is correct for all compliant deletion vectors (the fast path). A non-conforming writer could, however, store a compressed deletion vector blob; in that case the bytes at content_offset are compressed and the direct parse fails. Fall back to reading the blob through the Puffin footer, which records the per-blob compression-codec and decompresses accordingly (matching the blob by content_offset). Container-less blob files (no footer) stay on the fast path.

The fast path is covered by the existing round-trip test and real-data validation; the compressed fallback is defensive (the spec forbids compressed DVs, so no fixture is produced by conforming writers).

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,6 +31,7 @@ use crate::delete_vector::DeleteVector;
 use crate::expr::Predicate::AlwaysTrue;
 use crate::expr::{Predicate, Reference};
 use crate::io::FileIO;
+use crate::puffin::PuffinReader;
 use crate::runtime::Runtime;
 use crate::scan::{ArrowRecordBatchStream, FileScanTaskDeleteFile};
 use crate::spec::{
@@ -412,7 +413,45 @@ impl CachingDeleteFileLoader {
                     ),
                 )
             })?;
-        DeleteVector::from_serialized_bytes(&bytes[start..end])
+
+        // Fast path: per the Iceberg spec a `deletion-vector-v1` blob is stored
+        // uncompressed ("Omit compression-codec; deletion-vector-v1 is not
+        // compressed"), so the bytes located by content_offset/content_size are the
+        // serialized blob directly. This also covers container-less blob files
+        // (e.g. written by DuckDB) that have no Puffin footer.
+        if let Ok(delete_vector) = DeleteVector::from_serialized_bytes(&bytes[start..end]) {
+            return Ok(delete_vector);
+        }
+
+        // Fallback: a deletion vector blob that is compressed (which the spec
+        // forbids, but a non-conforming writer could produce) can only be
+        // decompressed using the codec recorded in the Puffin footer. If this file
+        // is a valid Puffin file, read the blob through its footer, which honors
+        // the per-blob compression codec.
+        let reader = PuffinReader::new(file_io.new_input(file_path)?);
+        let metadata = reader.file_metadata().await.map_err(|err| {
+            Error::new(
+                ErrorKind::DataInvalid,
+                format!(
+                    "deletion vector at offset {content_offset} in {file_path} is neither a valid uncompressed deletion-vector-v1 blob nor a Puffin container"
+                ),
+            )
+            .with_source(err)
+        })?;
+        let blob_metadata = metadata
+            .blobs()
+            .iter()
+            .find(|blob| blob.offset() == content_offset)
+            .ok_or_else(|| {
+                Error::new(
+                    ErrorKind::DataInvalid,
+                    format!(
+                        "Puffin file {file_path} has no blob at content_offset {content_offset}"
+                    ),
+                )
+            })?;
+        let blob = reader.blob(blob_metadata).await?;
+        DeleteVector::from_puffin_blob(blob)
     }
 
     /// Parses a record batch stream coming from positional delete files