fix(dv): reject trailing bitmap data

table/dv/deletion_vector_test.go

@@ -331,6 +331,23 @@ func TestDeserializeDVWrapsBitmapDecodeError(t *testing.T) {
 	assert.ErrorContains(t, err, "deserialize deletion vector bitmap")
 }
 
+// Why: a valid DV length and CRC must not hide extra bytes after the framed
+// roaring bitmap.
+// Condition: append bytes to a valid inner bitmap, then rebuild the envelope
+// so its length and CRC both match the malformed payload.
+// Assertion: returns a contextual bitmap error containing "trailing data".
+func TestDeserializeDVRejectsTrailingBitmapData(t *testing.T) {
+	bm := NewRoaringPositionBitmap()
+	bm.Set(7)
+	var bitmap bytes.Buffer
+	require.NoError(t, bm.Serialize(&bitmap))
+	bitmap.Write([]byte{0xde, 0xad})
+
+	_, err := DeserializeDV(wrapDVPayloadForTest(bitmap.Bytes()), -1)
+	assert.ErrorContains(t, err, "deserialize deletion vector bitmap")
+	assert.ErrorContains(t, err, "trailing data")
+}
+
 // Why: manifest metadata and DV content should agree on the number of deleted rows.
 // Condition: valid DV blob with 5 positions, but expected cardinality is set to 999.
 // Assertion: returns an error containing "cardinality mismatch".

table/dv/roaring_bitmap.go

@@ -179,6 +179,9 @@ func DeserializeRoaringPositionBitmap(data []byte) (*RoaringPositionBitmap, erro
 		lastKey = key
 		hasLastKey = true
 	}
+	if r.Len() != 0 {
+		return nil, fmt.Errorf("trailing data after bitmaps: %d bytes", r.Len())
+	}
 
 	return b, nil
 }

table/dv/roaring_bitmap_test.go

@@ -148,6 +148,33 @@ func TestDeserializeRoaringBitmapTruncatedAfterKey(t *testing.T) {
 	assert.ErrorContains(t, err, "read bitmap for key 0")
 }
 
+// Why: the outer count fully frames the portable bitmap and extra bytes are
+// malformed rather than a forward-compatible extension.
+// Condition: append bytes after otherwise valid empty and non-empty bitmaps.
+// Assertion: returns an error containing "trailing data".
+func TestDeserializeRoaringBitmapTrailingData(t *testing.T) {
+	for _, tt := range []struct {
+		name      string
+		positions []uint64
+	}{
+		{name: "empty"},
+		{name: "non-empty", positions: []uint64{1, (uint64(3) << 32) | 7}},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			bm := NewRoaringPositionBitmap()
+			for _, position := range tt.positions {
+				bm.Set(position)
+			}
+			var buf bytes.Buffer
+			require.NoError(t, bm.Serialize(&buf))
+			buf.Write([]byte{0xde, 0xad})
+
+			_, err := DeserializeRoaringPositionBitmap(buf.Bytes())
+			assert.ErrorContains(t, err, "trailing data")
+		})
+	}
+}
+
 // TestKeepMaskBytes pins KeepMaskBytes' contract: a bit-packed []byte where
 // bit i is 1 iff position i is NOT in the bitmap, sized to ⌈length/8⌉ bytes,
 // matching Arrow Boolean buffer layout so it can be wrapped without re-packing.