Core, Data: Validate deletion-vector offset and length

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

@@ -44,14 +44,15 @@
 import org.apache.iceberg.util.Pair;
 import org.apache.iceberg.util.Tasks;
 
-class DVUtil {
+public class DVUtil {
   private DVUtil() {}
 
   static PositionDeleteIndex readDV(DeleteFile deleteFile, FileIO fileIO) {
     Preconditions.checkArgument(
         ContentFileUtil.isDV(deleteFile),
         "Cannot read, not a deletion vector: %s",
         deleteFile.location());
+    validateDV(deleteFile);
     InputFile inputFile = fileIO.newInputFile(deleteFile);
     long offset = deleteFile.contentOffset();
     int length = deleteFile.contentSizeInBytes().intValue();
@@ -64,6 +65,38 @@ static PositionDeleteIndex readDV(DeleteFile deleteFile, FileIO fileIO) {
     }
   }
 
+  /**
+   * Validates that the deletion-vector offset and length on a {@link DeleteFile} are well-formed
+   * before they are consumed by a reader. Hostile or corrupted manifest metadata may otherwise
+   * trigger a {@link NegativeArraySizeException}, an invalid seek, or a multi-gigabyte allocation
+   * when the DV blob is read.
+   *
+   * @throws IllegalArgumentException if the offset or length is null or negative, or the length is
+   *     not less than 2GB
+   */
+  public static void validateDV(DeleteFile dv) {
+    Preconditions.checkArgument(
+        dv.contentOffset() != null,
+        "Invalid DV, offset cannot be null: %s",
+        ContentFileUtil.dvDesc(dv));
+    Preconditions.checkArgument(
+        dv.contentSizeInBytes() != null,
+        "Invalid DV, length cannot be null: %s",
+        ContentFileUtil.dvDesc(dv));
+    Preconditions.checkArgument(
+        dv.contentOffset() >= 0,
+        "Invalid DV, offset must be non-negative: %s",
+        ContentFileUtil.dvDesc(dv));
+    Preconditions.checkArgument(
+        dv.contentSizeInBytes() >= 0,
+        "Invalid DV, length must be non-negative: %s",
+        ContentFileUtil.dvDesc(dv));
+    Preconditions.checkArgument(
+        dv.contentSizeInBytes() < Integer.MAX_VALUE,
+        "Can't read DV larger than 2GB: %s",
+        dv.contentSizeInBytes());
+  }
+
   /**
    * Merges duplicate DVs for the same data file and writes the merged DV Puffin files. If there is
    * exactly 1 DV for a given data file then it is return as is

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

@@ -263,6 +263,16 @@ public DeleteFile build() {
       if (format == FileFormat.PUFFIN) {
         Preconditions.checkArgument(contentOffset != null, "Content offset is required for DV");
         Preconditions.checkArgument(contentSizeInBytes != null, "Content size is required for DV");
+        Preconditions.checkArgument(
+            contentOffset >= 0, "Content offset must be non-negative for DV: %s", contentOffset);
+        Preconditions.checkArgument(
+            contentSizeInBytes >= 0,
+            "Content size must be non-negative for DV: %s",
+            contentSizeInBytes);
+        Preconditions.checkArgument(
+            contentSizeInBytes < Integer.MAX_VALUE,
+            "Content size must be less than 2GB for DV: %s",
+            contentSizeInBytes);
         Preconditions.checkArgument(
             referencedDataFile != null, "Referenced data file is required for DV");
       } else {

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

@@ -0,0 +1,104 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg;
+
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import org.junit.jupiter.api.Test;
+
+public class TestDVUtil {
+
+  @Test
+  public void validateDVRejectsNullOffset() {
+    DeleteFile dv = dv(null, 10L);
+    assertThatThrownBy(() -> DVUtil.validateDV(dv))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("offset cannot be null");
+  }
+
+  @Test
+  public void validateDVRejectsNullLength() {
+    DeleteFile dv = dv(0L, null);
+    assertThatThrownBy(() -> DVUtil.validateDV(dv))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("length cannot be null");
+  }
+
+  @Test
+  public void validateDVRejectsNegativeOffset() {
+    DeleteFile dv = dv(-1L, 10L);
+    assertThatThrownBy(() -> DVUtil.validateDV(dv))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("offset must be non-negative");
+  }
+
+  @Test
+  public void validateDVRejectsNegativeLength() {
+    DeleteFile dv = dv(0L, -1L);
+    assertThatThrownBy(() -> DVUtil.validateDV(dv))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("length must be non-negative");
+  }
+
+  @Test
+  public void validateDVRejectsLengthEqualToIntegerMax() {
+    DeleteFile dv = dv(0L, (long) Integer.MAX_VALUE);
+    assertThatThrownBy(() -> DVUtil.validateDV(dv))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("Can't read DV larger than 2GB");
+  }
+
+  @Test
+  public void validateDVRejectsLengthAboveIntegerMax() {
+    DeleteFile dv = dv(0L, (long) Integer.MAX_VALUE + 1);
+    assertThatThrownBy(() -> DVUtil.validateDV(dv))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("Can't read DV larger than 2GB");
+  }
+
+  @Test
+  public void validateDVAcceptsZero() {
+    DeleteFile dv = dv(0L, 0L);
+    assertThatCode(() -> DVUtil.validateDV(dv)).doesNotThrowAnyException();
+  }
+
+  @Test
+  public void validateDVAcceptsTypicalValues() {
+    DeleteFile dv = dv(4L, 4096L);
+    assertThatCode(() -> DVUtil.validateDV(dv)).doesNotThrowAnyException();
+  }
+
+  @Test
+  public void validateDVAcceptsMaximumLength() {
+    DeleteFile dv = dv(0L, (long) Integer.MAX_VALUE - 1);
+    assertThatCode(() -> DVUtil.validateDV(dv)).doesNotThrowAnyException();
+  }
+
+  private static DeleteFile dv(Long offset, Long length) {
+    DeleteFile dv = mock(DeleteFile.class);
+    when(dv.location()).thenReturn("/tmp/test.puffin");
+    when(dv.referencedDataFile()).thenReturn("/tmp/data.parquet");
+    when(dv.contentOffset()).thenReturn(offset);
+    when(dv.contentSizeInBytes()).thenReturn(length);
+    return dv;
+  }
+}

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

@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import org.junit.jupiter.api.Test;
+
+public class TestFileMetadata {
+
+  @Test
+  public void dvBuilderRejectsNegativeContentOffset() {
+    assertThatThrownBy(() -> validDvBuilder().withContentOffset(-1L).build())
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("Content offset must be non-negative for DV");
+  }
+
+  @Test
+  public void dvBuilderRejectsNegativeContentSize() {
+    assertThatThrownBy(() -> validDvBuilder().withContentSizeInBytes(-1L).build())
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("Content size must be non-negative for DV");
+  }
+
+  @Test
+  public void dvBuilderRejectsContentSizeAtIntegerMax() {
+    assertThatThrownBy(() -> validDvBuilder().withContentSizeInBytes(Integer.MAX_VALUE).build())
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("less than 2GB");
+  }
+
+  @Test
+  public void dvBuilderAcceptsValidOffsetAndSize() {
+    DeleteFile dv = validDvBuilder().withContentOffset(4L).withContentSizeInBytes(4096L).build();
+
+    assertThat(dv.contentOffset()).isEqualTo(4L);
+    assertThat(dv.contentSizeInBytes()).isEqualTo(4096L);
+    assertThat(dv.format()).isEqualTo(FileFormat.PUFFIN);
+  }
+
+  @Test
+  public void dvBuilderAcceptsZeroOffsetAndSize() {
+    assertThatCode(() -> validDvBuilder().build()).doesNotThrowAnyException();
+  }
+
+  private static FileMetadata.Builder validDvBuilder() {
+    return FileMetadata.deleteFileBuilder(PartitionSpec.unpartitioned())
+        .ofPositionDeletes()
+        .withFormat(FileFormat.PUFFIN)
+        .withPath("/tmp/dv.puffin")
+        .withFileSizeInBytes(10)
+        .withRecordCount(1)
+        .withReferencedDataFile("/tmp/data.parquet")
+        .withContentOffset(0L)
+        .withContentSizeInBytes(0L);
+  }
+}

data/src/main/java/org/apache/iceberg/data/BaseDeleteLoader.java

@@ -25,6 +25,7 @@
 import java.util.concurrent.ExecutorService;
 import java.util.function.Function;
 import java.util.function.Supplier;
+import org.apache.iceberg.DVUtil;
 import org.apache.iceberg.DeleteFile;
 import org.apache.iceberg.FileFormat;
 import org.apache.iceberg.MetadataColumns;
@@ -264,18 +265,7 @@ private int estimateRecordSize(Schema schema) {
   }
 
   private void validateDV(DeleteFile dv, CharSequence filePath) {
-    Preconditions.checkArgument(
-        dv.contentOffset() != null,
-        "Invalid DV, offset cannot be null: %s",
-        ContentFileUtil.dvDesc(dv));
-    Preconditions.checkArgument(
-        dv.contentSizeInBytes() != null,
-        "Invalid DV, length is null: %s",
-        ContentFileUtil.dvDesc(dv));
-    Preconditions.checkArgument(
-        dv.contentSizeInBytes() <= Integer.MAX_VALUE,
-        "Can't read DV larger than 2GB: %s",
-        dv.contentSizeInBytes());
+    DVUtil.validateDV(dv);
     Preconditions.checkArgument(
         filePath.toString().equals(dv.referencedDataFile()),
         "DV is expected to reference %s, not %s",

data/src/test/java/org/apache/iceberg/data/TestBaseDeleteLoader.java

@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.data;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import org.apache.iceberg.DeleteFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.junit.jupiter.api.Test;
+
+public class TestBaseDeleteLoader {
+
+  private static final String DATA_FILE = "/tmp/data.parquet";
+
+  @Test
+  public void loadPositionDeletesRejectsNegativeContentOffset() {
+    DeleteFile dv = dv(-1L, 10L);
+    DeleteLoader loader = new BaseDeleteLoader(file -> failingInputFile());
+
+    assertThatThrownBy(() -> loader.loadPositionDeletes(ImmutableList.of(dv), DATA_FILE))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("offset must be non-negative");
+  }
+
+  @Test
+  public void loadPositionDeletesRejectsNegativeContentSize() {
+    DeleteFile dv = dv(0L, -1L);
+    DeleteLoader loader = new BaseDeleteLoader(file -> failingInputFile());
+
+    assertThatThrownBy(() -> loader.loadPositionDeletes(ImmutableList.of(dv), DATA_FILE))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessageContaining("length must be non-negative");
+  }
+
+  // Returns a DV mock reporting the given (offset, size). A mock bypasses the FileMetadata builder,
+  // which now rejects negative values at construction time, so it stands in for a corrupted
+  // manifest.
+  private static DeleteFile dv(Long offset, Long size) {
+    DeleteFile dv = mock(DeleteFile.class);
+    when(dv.format()).thenReturn(FileFormat.PUFFIN);
+    when(dv.location()).thenReturn("/tmp/dv.puffin");
+    when(dv.referencedDataFile()).thenReturn(DATA_FILE);
+    when(dv.contentOffset()).thenReturn(offset);
+    when(dv.contentSizeInBytes()).thenReturn(size);
+    return dv;
+  }
+
+  // Validation must fire before any I/O. If it does not, the loader calls this and the test fails
+  // with a clear message instead of an obscure stack trace.
+  private static InputFile failingInputFile() {
+    throw new AssertionError("DV validation should reject the invalid metadata before any I/O");
+  }
+}