diff --git a/table/dv/deletion_vector.go b/table/dv/deletion_vector.go index 7eb378a7b..0e10c78d5 100644 --- a/table/dv/deletion_vector.go +++ b/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 } diff --git a/table/dv/deletion_vector_test.go b/table/dv/deletion_vector_test.go index 945447fb4..022a834ed 100644 --- a/table/dv/deletion_vector_test.go +++ b/table/dv/deletion_vector_test.go @@ -139,20 +139,23 @@ 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}}) @@ -160,7 +163,7 @@ func writeRawPuffinWithDVBlobNoCardinality(t *testing.T, dir string, dvBlobBytes 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