fix(dv): validate Puffin blob metadata against manifest

table/dv/deletion_vector.go

@@ -35,7 +35,10 @@ import (
 // writer attaches it; the reader uses it to detect truncated bitmaps whose
 // CRC happens to validate (the CRC covers the bytes that ARE present, not
 // the bytes that should have been).
-const dvCardinalityProperty = "cardinality"
+const (
+	dvCardinalityProperty        = "cardinality"
+	dvReferencedDataFileProperty = "referenced-data-file"
+)
 
 const (
 	// DVMagicNumber is the magic number for deletion vectors.
@@ -158,6 +161,10 @@ func ReadDV(fs iceio.IO, dvFile iceberg.DataFile) (*RoaringPositionBitmap, error
 	if dvFile.ContentOffset() == nil || dvFile.ContentSizeInBytes() == nil {
 		return nil, fmt.Errorf("DV file %s missing ContentOffset/ContentSizeInBytes", dvFile.FilePath())
 	}
+	manifestReferencedDataFile := dvFile.ReferencedDataFile()
+	if manifestReferencedDataFile == nil || *manifestReferencedDataFile == "" {
+		return nil, fmt.Errorf("DV file %s missing ReferencedDataFile", dvFile.FilePath())
+	}
 
 	size := *dvFile.ContentSizeInBytes()
 	if size < 0 || size > int64(puffin.DefaultMaxBlobSize) {
@@ -176,9 +183,23 @@ func ReadDV(fs iceio.IO, dvFile iceberg.DataFile) (*RoaringPositionBitmap, error
 	}
 
 	offset := *dvFile.ContentOffset()
-	blobData := make([]byte, size)
-	if _, err := reader.ReadAt(blobData, offset); err != nil {
-		return nil, fmt.Errorf("read DV blob at offset %d: %w", offset, err)
+	blob, err := findBlobMetadataByRange(reader.Blobs(), offset, size)
+	if err != nil {
+		return nil, fmt.Errorf("DV file %s: %w", dvFile.FilePath(), err)
+	}
+	if blob.Type != puffin.BlobTypeDeletionVector {
+		return nil, fmt.Errorf("DV file %s: blob at offset %d has type %q, expected %q",
+			dvFile.FilePath(), offset, blob.Type, puffin.BlobTypeDeletionVector)
+	}
+
+	referencedDataFile, ok := blob.Properties[dvReferencedDataFileProperty]
+	if !ok || referencedDataFile == "" {
+		return nil, fmt.Errorf("DV file %s: blob at offset %d missing %s property",
+			dvFile.FilePath(), offset, dvReferencedDataFileProperty)
+	}
+	if referencedDataFile != *manifestReferencedDataFile {
+		return nil, fmt.Errorf("DV file %s: manifest referenced_data_file %q does not match puffin %s %q",
+			dvFile.FilePath(), *manifestReferencedDataFile, dvReferencedDataFileProperty, referencedDataFile)
 	}
 
 	// Manifest record_count (field 103) is a required, non-nullable long, so it
@@ -188,7 +209,7 @@ func ReadDV(fs iceio.IO, dvFile iceberg.DataFile) (*RoaringPositionBitmap, error
 
 	// The puffin blob independently declares its cardinality via the spec-
 	// mandated property; when present it is a second source to cross-check.
-	puffinCardinality, hasPuffinCardinality, err := blobCardinality(reader.Blobs(), offset, size)
+	puffinCardinality, hasPuffinCardinality, err := blobCardinality(blob)
 	if err != nil {
 		return nil, fmt.Errorf("DV file %s: %w", dvFile.FilePath(), err)
 	}
@@ -211,23 +232,20 @@ func ReadDV(fs iceio.IO, dvFile iceberg.DataFile) (*RoaringPositionBitmap, error
 			"dv_file", dvFile.FilePath(), "offset", offset)
 	}
 
+	blobData := make([]byte, size)
+	if _, err := reader.ReadAt(blobData, offset); err != nil {
+		return nil, fmt.Errorf("read DV blob at offset %d: %w", offset, err)
+	}
+
 	// Validate the decoded bitmap against the manifest record_count (always
 	// present, including zero). When the puffin property is present it has
 	// already been confirmed to agree with this value above.
 	return DeserializeDV(blobData, manifestCardinality)
 }
 
-// blobCardinality returns the cardinality declared by the puffin blob at the
-// manifest entry's (offset, size). The bool indicates whether the property was
-// present:
-//
-//   - (n, true, nil)  — property found and parsed successfully
-//   - (0, false, nil) — matching blob found but no cardinality property
-//   - (_, _, err)     — manifest/footer mismatch or property unparseable
-//
-// Keeping the sentinel out of the int64 return channel avoids leaking
-// DeserializeDV's "-1 means skip" convention up the call chain.
-func blobCardinality(blobs []puffin.BlobMetadata, offset, size int64) (int64, bool, error) {
+// findBlobMetadataByRange returns the footer entry identified by the
+// manifest's content offset and size.
+func findBlobMetadataByRange(blobs []puffin.BlobMetadata, offset, size int64) (puffin.BlobMetadata, error) {
 	for _, b := range blobs {
 		if b.Offset != offset {
 			continue
@@ -237,26 +255,31 @@ func blobCardinality(blobs []puffin.BlobMetadata, offset, size int64) (int64, bo
 			// disagrees with the puffin footer on how big this blob is.
 			// Surface that distinct condition rather than rolling it into
 			// "no blob at offset" — different writer bug, different fix.
-			return 0, false, fmt.Errorf("blob at offset %d has length %d, manifest says %d", offset, b.Length, size)
-		}
-		v, ok := b.Properties[dvCardinalityProperty]
-		if !ok {
-			return 0, false, nil
-		}
-		parsed, err := strconv.ParseInt(v, 10, 64)
-		if err != nil {
-			return 0, false, fmt.Errorf("invalid %s property %q: %w", dvCardinalityProperty, v, err)
-		}
-		if parsed < 0 {
-			// Negative is meaningless for a count of deleted positions, and
-			// `-1` specifically aliases DeserializeDV's skip-validation
-			// sentinel — accepting it would silently disable the very check
-			// this layer was added to perform.
-			return 0, false, fmt.Errorf("%s property must be non-negative, got %d", dvCardinalityProperty, parsed)
+			return puffin.BlobMetadata{}, fmt.Errorf("blob at offset %d has length %d, manifest says %d", offset, b.Length, size)
 		}
 
-		return parsed, true, nil
+		return b, nil
+	}
+
+	return puffin.BlobMetadata{}, fmt.Errorf("no blob in puffin footer at offset %d, size %d", offset, size)
+}
+
+// blobCardinality returns the cardinality declared by a Puffin blob. The bool
+// indicates whether the property was present.
+func blobCardinality(blob puffin.BlobMetadata) (int64, bool, error) {
+	v, ok := blob.Properties[dvCardinalityProperty]
+	if !ok {
+		return 0, false, nil
+	}
+	parsed, err := strconv.ParseInt(v, 10, 64)
+	if err != nil {
+		return 0, false, fmt.Errorf("invalid %s property %q: %w", dvCardinalityProperty, v, err)
+	}
+	if parsed < 0 {
+		// Negative is meaningless for a count of deleted positions, and -1
+		// would disable cardinality validation in DeserializeDV.
+		return 0, false, fmt.Errorf("%s property must be non-negative, got %d", dvCardinalityProperty, parsed)
 	}
 
-	return 0, false, fmt.Errorf("no blob in puffin footer at offset %d, size %d", offset, size)
+	return parsed, true, nil
 }

table/dv/deletion_vector_test.go

@@ -139,28 +139,31 @@ func writePuffinWithDVBlobAndProps(t *testing.T, dir string, dvBlobBytes []byte,
 // writer) is to assemble the bytes directly. The reader's validateBlobs does
 // not require the property, so this file loads cleanly.
 func writeRawPuffinWithDVBlobNoCardinality(t *testing.T, dir string, dvBlobBytes []byte) (string, puffin.BlobMetadata) {
+	return writeRawPuffinBlob(t, dir, "raw-dv-no-cardinality.puffin", dvBlobBytes, puffin.BlobTypeDeletionVector, map[string]string{
+		"referenced-data-file": "s3://bucket/data/data-001.parquet",
+	})
+}
+
+func writeRawPuffinBlob(t *testing.T, dir, name string, blobBytes []byte, blobType puffin.BlobType, props map[string]string) (string, puffin.BlobMetadata) {
 	t.Helper()
 
 	const puffinMagic = "PFA1"
 	meta := puffin.BlobMetadata{
-		Type:           puffin.BlobTypeDeletionVector,
+		Type:           blobType,
 		Fields:         []int32{2147483546},
 		SnapshotID:     -1,
 		SequenceNumber: -1,
 		Offset:         int64(puffin.MagicSize),
-		Length:         int64(len(dvBlobBytes)),
-		Properties: map[string]string{
-			// No "cardinality" — that is the whole point of this fixture.
-			"referenced-data-file": "s3://bucket/data/data-001.parquet",
-		},
+		Length:         int64(len(blobBytes)),
+		Properties:     props,
 	}
 
 	payload, err := json.Marshal(puffin.Footer{Blobs: []puffin.BlobMetadata{meta}})
 	require.NoError(t, err)
 
 	var buf bytes.Buffer
 	buf.WriteString(puffinMagic) // header magic
-	buf.Write(dvBlobBytes)       // blob at offset MagicSize
+	buf.Write(blobBytes)         // blob at offset MagicSize
 	buf.WriteString(puffinMagic) // footer start magic
 	buf.Write(payload)           // footer JSON payload
 
@@ -170,7 +173,7 @@ func writeRawPuffinWithDVBlobNoCardinality(t *testing.T, dir string, dvBlobBytes
 	copy(trailer[8:12], puffinMagic)
 	buf.Write(trailer[:])
 
-	path := filepath.Join(dir, "raw-dv-no-cardinality.puffin")
+	path := filepath.Join(dir, name)
 	require.NoError(t, os.WriteFile(path, buf.Bytes(), 0o644))
 
 	return path, meta
@@ -390,6 +393,28 @@ func TestReadDVMissingContentMetadata(t *testing.T) {
 	})
 }
 
+// Why: a deletion vector manifest entry must identify the data file it applies to.
+// Condition: ReferencedDataFile is nil or empty even though offset and size are present.
+// Assertion: ReadDV rejects the entry before opening the Puffin file.
+func TestReadDVMissingReferencedDataFile(t *testing.T) {
+	for _, tt := range []struct {
+		name       string
+		referenced *string
+	}{
+		{name: "nil", referenced: nil},
+		{name: "empty", referenced: strPtr("")},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			offset, size := int64(4), int64(50)
+			dvFile := newDVTestFile("missing.puffin", 5, &offset, &size)
+			dvFile.referencedDataFile = tt.referenced
+
+			_, err := ReadDV(iceio.LocalFS{}, dvFile)
+			assert.ErrorContains(t, err, "missing ReferencedDataFile")
+		})
+	}
+}
+
 // Why: negative or absurdly large blob sizes should be rejected before allocation.
 // Condition: ContentSizeInBytes set to -1.
 // Assertion: returns an error containing "out of valid range".
@@ -425,9 +450,55 @@ func TestReadDVInvalidPuffin(t *testing.T) {
 	assert.ErrorContains(t, err, "create puffin reader")
 }
 
+// Why: offset, size, and cardinality cannot prove that the selected Puffin blob
+// is a deletion vector for the manifest's referenced data file.
+// Condition: the matched blob has conflicting, missing, or non-DV identity metadata.
+// Assertion: ReadDV rejects each case before decoding the blob payload.
+func TestReadDVValidatesBlobMetadata(t *testing.T) {
+	dvBlobBytes := readDVTestData(t, "small-alternating-values-position-index.bin")
+
+	t.Run("mismatched referenced data file", func(t *testing.T) {
+		dir := t.TempDir()
+		path, meta := writePuffinWithDVBlobAndProps(t, dir, dvBlobBytes, map[string]string{
+			"referenced-data-file": "s3://bucket/data/data-002.parquet",
+			"cardinality":          "5",
+		})
+		offset, size := meta.Offset, meta.Length
+
+		_, err := ReadDV(iceio.LocalFS{}, newDVTestFile(path, 5, &offset, &size))
+		require.Error(t, err)
+		assert.ErrorContains(t, err, "manifest referenced_data_file")
+		assert.ErrorContains(t, err, "data-001.parquet")
+		assert.ErrorContains(t, err, "data-002.parquet")
+	})
+
+	t.Run("missing puffin referenced data file", func(t *testing.T) {
+		dir := t.TempDir()
+		path, meta := writeRawPuffinBlob(t, dir, "raw-dv-no-reference.puffin", dvBlobBytes,
+			puffin.BlobTypeDeletionVector, map[string]string{"cardinality": "5"})
+		offset, size := meta.Offset, meta.Length
+
+		_, err := ReadDV(iceio.LocalFS{}, newDVTestFile(path, 5, &offset, &size))
+		assert.ErrorContains(t, err, "missing referenced-data-file property")
+	})
+
+	t.Run("wrong blob type", func(t *testing.T) {
+		dir := t.TempDir()
+		path, meta := writeRawPuffinBlob(t, dir, "raw-wrong-type.puffin", dvBlobBytes,
+			puffin.BlobType("test-blob"), map[string]string{
+				"referenced-data-file": "s3://bucket/data/data-001.parquet",
+				"cardinality":          "5",
+			})
+		offset, size := meta.Offset, meta.Length
+
+		_, err := ReadDV(iceio.LocalFS{}, newDVTestFile(path, 5, &offset, &size))
+		assert.ErrorContains(t, err, `has type "test-blob", expected "deletion-vector-v1"`)
+	})
+}
+
 // Why: manifest-provided blob ranges must point into the Puffin blob area, not arbitrary offsets.
 // Condition: valid Puffin file, but content offset is forced to 0, which points before the blob region.
-// Assertion: returns an error containing "read DV blob at offset 0".
+// Assertion: returns an error indicating that no footer blob matches the range.
 func TestReadDVInvalidBlobRange(t *testing.T) {
 	dvBlobBytes := readDVTestData(t, "small-alternating-values-position-index.bin")
 
@@ -436,7 +507,7 @@ func TestReadDVInvalidBlobRange(t *testing.T) {
 
 	offset, size := int64(0), meta.Length
 	_, err := ReadDV(iceio.LocalFS{}, newDVTestFile(path, 5, &offset, &size))
-	assert.ErrorContains(t, err, "read DV blob at offset 0")
+	assert.ErrorContains(t, err, "no blob in puffin footer at offset 0")
 }
 
 // TestReadDVCardinalityValidation pins the spec-mandated check that the