Core: Include CRC values in DV deserialization error message

core/src/main/java/org/apache/iceberg/deletes/BitmapPositionDeleteIndex.java

@@ -150,7 +150,12 @@ public static PositionDeleteIndex deserialize(byte[] bytes, DeleteFile deleteFil
     int crc = computeChecksum(bytes, bitmapDataLength);
     int crcOffset = LENGTH_SIZE_BYTES + bitmapDataLength;
     int expectedCrc = buffer.getInt(crcOffset);
-    Preconditions.checkArgument(crc == expectedCrc, "Invalid CRC");
+    Preconditions.checkArgument(
+        crc == expectedCrc,
+        "Invalid CRC for deletion vector %s: 0x%s, expected 0x%s",
+        deleteFile.location(),
+        Integer.toHexString(crc),
+        Integer.toHexString(expectedCrc));
     return new BitmapPositionDeleteIndex(bitmap, deleteFile);
   }
 

core/src/test/java/org/apache/iceberg/deletes/TestBitmapPositionDeleteIndex.java

@@ -19,6 +19,7 @@
 package org.apache.iceberg.deletes;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import java.io.IOException;
 import java.net.URL;
@@ -176,6 +177,24 @@ public void testAllContainerTypesIndexSerialization() throws Exception {
     validate(index, "all-container-types-position-index.bin");
   }
 
+  @Test
+  public void testDeserializeInvalidCrc() {
+    PositionDeleteIndex index = new BitmapPositionDeleteIndex();
+    index.delete(1L);
+    index.delete(2L);
+    byte[] bytes = index.serialize().array();
+
+    // corrupt the last CRC byte so the computed checksum no longer matches
+    bytes[bytes.length - 1] ^= 0x01;
+
+    DeleteFile dv = mockDV(bytes.length, index.cardinality());
+    Mockito.when(dv.location()).thenReturn("s3://bucket/dv.puffin");
+
+    assertThatThrownBy(() -> PositionDeleteIndex.deserialize(bytes, dv))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("Invalid CRC for deletion vector s3://bucket/dv.puffin");
+  }
+
   private static void validate(PositionDeleteIndex index, String goldenFile) throws Exception {
     ByteBuffer buffer = index.serialize();
     byte[] bytes = buffer.array();