Core: Extend V4 DeletionVector with key_metadata field

core/src/main/java/org/apache/iceberg/DeletionVector.java

@@ -18,6 +18,7 @@
  */
 package org.apache.iceberg;
 
+import java.nio.ByteBuffer;
 import org.apache.iceberg.types.Types;
 
 /**
@@ -45,9 +46,15 @@ interface DeletionVector {
           "cardinality",
           Types.LongType.get(),
           "Number of set bits (deleted rows) in the vector");
+  Types.NestedField KEY_METADATA =
+      Types.NestedField.optional(
+          149,
+          "key_metadata",
+          Types.BinaryType.get(),
+          "Implementation-specific key metadata for encryption");
 
   static Types.StructType schema() {
-    return Types.StructType.of(LOCATION, OFFSET, SIZE_IN_BYTES, CARDINALITY);
+    return Types.StructType.of(LOCATION, OFFSET, SIZE_IN_BYTES, CARDINALITY, KEY_METADATA);
   }
 
   /** Returns the location of the file containing the deletion vector. */
@@ -62,6 +69,9 @@ static Types.StructType schema() {
   /** Returns the number of set bits (deleted rows) in the vector. */
   long cardinality();
 
+  /** Returns encryption key metadata, or null if the file is not encrypted. */
+  ByteBuffer keyMetadata();
+
   /** Copies this deletion vector. */
   DeletionVector copy();
 }

core/src/main/java/org/apache/iceberg/DeletionVectorStruct.java

@@ -19,11 +19,14 @@
 package org.apache.iceberg;
 
 import java.io.Serializable;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
 import java.util.Objects;
 import org.apache.iceberg.avro.SupportsIndexProjection;
 import org.apache.iceberg.relocated.com.google.common.base.MoreObjects;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.ByteBuffers;
 
 /** Mutable {@link StructLike} implementation of {@link DeletionVector}. */
 class DeletionVectorStruct extends SupportsIndexProjection implements DeletionVector, Serializable {
@@ -32,12 +35,14 @@ class DeletionVectorStruct extends SupportsIndexProjection implements DeletionVe
           DeletionVector.LOCATION,
           DeletionVector.OFFSET,
           DeletionVector.SIZE_IN_BYTES,
-          DeletionVector.CARDINALITY);
+          DeletionVector.CARDINALITY,
+          DeletionVector.KEY_METADATA);
 
   private String location = null;
   private long offset = -1L;
   private long sizeInBytes = -1L;
   private long cardinality = -1L;
+  private byte[] keyMetadata = null;
 
   DeletionVectorStruct(Types.StructType type) {
     super(BASE_TYPE, type);
@@ -49,14 +54,20 @@ private DeletionVectorStruct(DeletionVectorStruct toCopy) {
     this.offset = toCopy.offset;
     this.sizeInBytes = toCopy.sizeInBytes;
     this.cardinality = toCopy.cardinality;
+    this.keyMetadata =
+        toCopy.keyMetadata != null
+            ? Arrays.copyOf(toCopy.keyMetadata, toCopy.keyMetadata.length)
+            : null;
   }
 
-  private DeletionVectorStruct(String location, long offset, long sizeInBytes, long cardinality) {
+  private DeletionVectorStruct(
+      String location, long offset, long sizeInBytes, long cardinality, ByteBuffer keyMetadata) {
     super(BASE_TYPE.fields().size());
     this.location = location;
     this.offset = offset;
     this.sizeInBytes = sizeInBytes;
     this.cardinality = cardinality;
+    this.keyMetadata = ByteBuffers.toByteArray(keyMetadata);
   }
 
   @Override
@@ -79,6 +90,11 @@ public long cardinality() {
     return cardinality;
   }
 
+  @Override
+  public ByteBuffer keyMetadata() {
+    return keyMetadata != null ? ByteBuffer.wrap(keyMetadata) : null;
+  }
+
   @Override
   public DeletionVectorStruct copy() {
     return new DeletionVectorStruct(this);
@@ -90,38 +106,28 @@ protected <T> T internalGet(int pos, Class<T> javaClass) {
   }
 
   private Object getByPos(int pos) {
-    switch (pos) {
-      case 0:
-        return location;
-      case 1:
-        return offset;
-      case 2:
-        return sizeInBytes;
-      case 3:
-        return cardinality;
-      default:
-        throw new UnsupportedOperationException("Unknown field ordinal: " + pos);
-    }
+    return switch (pos) {
+      case 0 -> location;
+      case 1 -> offset;
+      case 2 -> sizeInBytes;
+      case 3 -> cardinality;
+      case 4 -> keyMetadata();
+      default -> throw new UnsupportedOperationException("Unknown field ordinal: " + pos);
+    };
   }
 
   @Override
   protected <T> void internalSet(int pos, T value) {
     switch (pos) {
-      case 0:
         // always coerce to String for Serializable
-        this.location = value.toString();
-        break;
-      case 1:
-        this.offset = (Long) value;
-        break;
-      case 2:
-        this.sizeInBytes = (Long) value;
-        break;
-      case 3:
-        this.cardinality = (Long) value;
-        break;
-      default:
+      case 0 -> this.location = value.toString();
+      case 1 -> this.offset = (Long) value;
+      case 2 -> this.sizeInBytes = (Long) value;
+      case 3 -> this.cardinality = (Long) value;
+      case 4 -> this.keyMetadata = ByteBuffers.toByteArray((ByteBuffer) value);
+      default -> {
         // ignore the object, it must be from a newer version of the format
+      }
     }
   }
 
@@ -156,6 +162,7 @@ public String toString() {
         .add("offset", offset)
         .add("size_in_bytes", sizeInBytes)
         .add("cardinality", cardinality)
+        .add("key_metadata", keyMetadata == null ? "null" : "(redacted)")
         .toString();
   }
 
@@ -164,6 +171,7 @@ static class Builder {
     private Long offset = null;
     private Long sizeInBytes = null;
     private Long cardinality = null;
+    private ByteBuffer keyMetadata = null;
 
     Builder location(String dvLocation) {
       Preconditions.checkArgument(dvLocation != null, "Invalid location: null");
@@ -191,12 +199,17 @@ Builder cardinality(long dvCardinality) {
       return this;
     }
 
+    Builder keyMetadata(ByteBuffer dvKeyMetadata) {
+      this.keyMetadata = dvKeyMetadata;
+      return this;
+    }
+
     DeletionVectorStruct build() {
       Preconditions.checkArgument(location != null, "Missing required value: location");
       Preconditions.checkArgument(offset != null, "Missing required value: offset");
       Preconditions.checkArgument(sizeInBytes != null, "Missing required value: size in bytes");
       Preconditions.checkArgument(cardinality != null, "Missing required value: cardinality");
-      return new DeletionVectorStruct(location, offset, sizeInBytes, cardinality);
+      return new DeletionVectorStruct(location, offset, sizeInBytes, cardinality, keyMetadata);
     }
   }
 }

core/src/main/java/org/apache/iceberg/TrackedFileAdapters.java

@@ -331,7 +331,7 @@ public Integer sortOrderId() {
 
     @Override
     public ByteBuffer keyMetadata() {
-      return null;
+      return dv.keyMetadata();
     }
 
     @Override

core/src/test/java/org/apache/iceberg/TestDeletionVectorStruct.java

@@ -22,11 +22,14 @@
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import java.io.IOException;
+import java.nio.ByteBuffer;
 import org.apache.iceberg.types.Types;
 import org.junit.jupiter.api.Test;
 
 class TestDeletionVectorStruct {
 
+  private static final ByteBuffer KEY_METADATA = ByteBuffer.wrap(new byte[] {1, 2, 3});
+
   @Test
   void testFieldAccess() {
     DeletionVectorStruct dv =
@@ -35,12 +38,14 @@ void testFieldAccess() {
             .offset(256L)
             .sizeInBytes(128L)
             .cardinality(42L)
+            .keyMetadata(KEY_METADATA)
             .build();
 
     assertThat(dv.location()).isEqualTo("s3://bucket/data/dv.puffin");
     assertThat(dv.offset()).isEqualTo(256L);
     assertThat(dv.sizeInBytes()).isEqualTo(128L);
     assertThat(dv.cardinality()).isEqualTo(42L);
+    assertThat(dv.keyMetadata()).isEqualTo(KEY_METADATA);
   }
 
   @Test
@@ -51,6 +56,7 @@ void testCopy() {
             .offset(256L)
             .sizeInBytes(128L)
             .cardinality(42L)
+            .keyMetadata(KEY_METADATA)
             .build();
 
     DeletionVectorStruct copy = dv.copy();
@@ -59,12 +65,13 @@ void testCopy() {
     assertThat(copy.offset()).isEqualTo(256L);
     assertThat(copy.sizeInBytes()).isEqualTo(128L);
     assertThat(copy.cardinality()).isEqualTo(42L);
+    assertThat(copy.keyMetadata()).isEqualTo(KEY_METADATA);
   }
 
   @Test
   void testSize() {
     DeletionVectorStruct dv = new DeletionVectorStruct(DeletionVector.schema());
-    assertThat(dv.size()).isEqualTo(4);
+    assertThat(dv.size()).isEqualTo(5);
   }
 
   @Test
@@ -95,6 +102,7 @@ void testInternalSetIgnoresUnknownOrdinal() {
             .offset(100L)
             .sizeInBytes(512L)
             .cardinality(42L)
+            .keyMetadata(KEY_METADATA)
             .build();
 
     // unknown ordinals from a newer format version are silently ignored
@@ -105,6 +113,7 @@ void testInternalSetIgnoresUnknownOrdinal() {
     assertThat(dv.offset()).isEqualTo(100L);
     assertThat(dv.sizeInBytes()).isEqualTo(512L);
     assertThat(dv.cardinality()).isEqualTo(42L);
+    assertThat(dv.keyMetadata()).isEqualTo(KEY_METADATA);
   }
 
   @Test
@@ -115,6 +124,7 @@ void testJavaSerializationRoundTrip() throws IOException, ClassNotFoundException
             .offset(256L)
             .sizeInBytes(128L)
             .cardinality(42L)
+            .keyMetadata(KEY_METADATA)
             .build();
 
     DeletionVectorStruct deserialized = TestHelpers.roundTripSerialize(dv);
@@ -123,6 +133,7 @@ void testJavaSerializationRoundTrip() throws IOException, ClassNotFoundException
     assertThat(deserialized.offset()).isEqualTo(256L);
     assertThat(deserialized.sizeInBytes()).isEqualTo(128L);
     assertThat(deserialized.cardinality()).isEqualTo(42L);
+    assertThat(deserialized.keyMetadata()).isEqualTo(KEY_METADATA);
   }
 
   @Test
@@ -248,6 +259,7 @@ void testKryoSerializationRoundTrip() throws IOException {
             .offset(256L)
             .sizeInBytes(128L)
             .cardinality(42L)
+            .keyMetadata(KEY_METADATA)
             .build();
 
     DeletionVectorStruct deserialized = TestHelpers.KryoHelpers.roundTripSerialize(dv);
@@ -256,5 +268,6 @@ void testKryoSerializationRoundTrip() throws IOException {
     assertThat(deserialized.offset()).isEqualTo(256L);
     assertThat(deserialized.sizeInBytes()).isEqualTo(128L);
     assertThat(deserialized.cardinality()).isEqualTo(42L);
+    assertThat(deserialized.keyMetadata()).isEqualTo(KEY_METADATA);
   }
 }

core/src/test/java/org/apache/iceberg/TestTrackedFileAdapters.java

@@ -49,6 +49,8 @@ class TestTrackedFileAdapters {
   private static final Map<Integer, PartitionSpec> UNPARTITIONED =
       ImmutableMap.of(UNPARTITIONED_SPEC_ID, PartitionSpec.unpartitioned());
 
+  private static final ByteBuffer KEY_METADATA = ByteBuffer.wrap(new byte[] {1, 2, 3});
+
   private static final Schema PARTITION_SCHEMA =
       new Schema(Types.NestedField.required(1, "category", Types.StringType.get()));
   private static final int PARTITIONED_SPEC_ID = 1;
@@ -249,6 +251,7 @@ void testDVDeleteFileAdapterDelegation() {
             .offset(128L)
             .sizeInBytes(256L)
             .cardinality(10L)
+            .keyMetadata(KEY_METADATA)
             .build();
 
     TrackingStruct tracking =
@@ -292,6 +295,7 @@ void testDVDeleteFileAdapterDelegation() {
     assertThat(dvFile.recordCount()).isEqualTo(dv.cardinality());
     assertThat(dvFile.contentOffset()).isEqualTo(dv.offset());
     assertThat(dvFile.contentSizeInBytes()).isEqualTo(dv.sizeInBytes());
+    assertThat(dvFile.keyMetadata()).isEqualTo(KEY_METADATA);
     // fileSizeInBytes reports the DV blob size, not the full Puffin file size.
     assertThat(dvFile.fileSizeInBytes()).isEqualTo(dv.sizeInBytes());
     // referencedDataFile is delegated to the tracked data file's location.
@@ -308,7 +312,6 @@ void testDVDeleteFileAdapterDelegation() {
     // fields that are null for DVs
     assertThat(dvFile.sortOrderId()).isNull();
     assertThat(dvFile.firstRowId()).isNull();
-    assertThat(dvFile.keyMetadata()).isNull();
     assertThat(dvFile.splitOffsets()).isNull();
     assertThat(dvFile.equalityFieldIds()).isNull();
     assertThat(dvFile.columnSizes()).isNull();