From dd3384a365a5ab12518afc33f17c1401b6c19272 Mon Sep 17 00:00:00 2001
From: xuanyili <xuanyili@uber.com>
Date: Sat, 5 Sep 2026 17:19:33 +0000
Subject: [PATCH 1/4] test: add cross-language deletion vector fixtures

---
 dev/dv-fixtures/generate_go_fixtures.go       | 152 +++++++++++
 src/iceberg/test/CMakeLists.txt               |   1 +
 src/iceberg/test/puffin_dv_interop_test.cc    | 256 ++++++++++++++++++
 .../test/resources/deletion_vectors/README.md |  44 +++
 .../go/all-container-types-dv.puffin          | Bin 0 -> 16781 bytes
 .../deletion_vectors/go/multi-blob-dv.puffin  | Bin 0 -> 650 bytes
 .../deletion_vectors/go/single-blob-dv.puffin | Bin 0 -> 314 bytes
 .../java/multi-blob-dv.puffin                 | Bin 0 -> 639 bytes
 .../java/single-blob-dv.puffin                | Bin 0 -> 384 bytes
 9 files changed, 453 insertions(+)
 create mode 100644 dev/dv-fixtures/generate_go_fixtures.go
 create mode 100644 src/iceberg/test/puffin_dv_interop_test.cc
 create mode 100644 src/iceberg/test/resources/deletion_vectors/README.md
 create mode 100644 src/iceberg/test/resources/deletion_vectors/go/all-container-types-dv.puffin
 create mode 100644 src/iceberg/test/resources/deletion_vectors/go/multi-blob-dv.puffin
 create mode 100644 src/iceberg/test/resources/deletion_vectors/go/single-blob-dv.puffin
 create mode 100644 src/iceberg/test/resources/deletion_vectors/java/multi-blob-dv.puffin
 create mode 100644 src/iceberg/test/resources/deletion_vectors/java/single-blob-dv.puffin

diff --git a/dev/dv-fixtures/generate_go_fixtures.go b/dev/dv-fixtures/generate_go_fixtures.go
new file mode 100644
index 000000000..67d26d6c0
--- /dev/null
+++ b/dev/dv-fixtures/generate_go_fixtures.go
@@ -0,0 +1,152 @@
+// 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.
+
+//go:build ignore
+
+package main
+
+import (
+	"bytes"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strconv"
+
+	"github.com/apache/iceberg-go/puffin"
+	"github.com/apache/iceberg-go/table/dv"
+)
+
+type fixtureBlob struct {
+	referencedDataFile string
+	positions          []uint64
+	ranges             []positionRange
+}
+
+type positionRange struct {
+	start uint64
+	end   uint64
+}
+
+func writeFixture(outputDir, fileName, createdBy string, blobs []fixtureBlob) error {
+	var output bytes.Buffer
+	writer, err := puffin.NewWriter(&output)
+	if err != nil {
+		return err
+	}
+	if err := writer.SetCreatedBy(createdBy); err != nil {
+		return err
+	}
+
+	for _, blob := range blobs {
+		bitmap := dv.NewRoaringPositionBitmap()
+		for _, position := range blob.positions {
+			bitmap.Set(position)
+		}
+		for _, positionRange := range blob.ranges {
+			bitmap.SetRange(positionRange.start, positionRange.end)
+		}
+		payload, err := dv.SerializeDV(bitmap)
+		if err != nil {
+			return err
+		}
+		_, err = writer.AddBlob(puffin.BlobMetadataInput{
+			Type:           puffin.BlobTypeDeletionVector,
+			SnapshotID:     -1,
+			SequenceNumber: -1,
+			Fields:         []int32{},
+			Properties: map[string]string{
+				"referenced-data-file": blob.referencedDataFile,
+				"cardinality":          strconv.FormatInt(bitmap.Cardinality(), 10),
+			},
+		}, payload)
+		if err != nil {
+			return err
+		}
+	}
+
+	if err := writer.Finish(); err != nil {
+		return err
+	}
+	return os.WriteFile(filepath.Join(outputDir, fileName), output.Bytes(), 0o644)
+}
+
+func main() {
+	if len(os.Args) != 2 {
+		fmt.Fprintln(os.Stderr, "usage: go run generate_go_fixtures.go OUTPUT_DIR")
+		os.Exit(2)
+	}
+	outputDir := os.Args[1]
+	if err := os.MkdirAll(outputDir, 0o755); err != nil {
+		panic(err)
+	}
+
+	err := writeFixture(outputDir, "single-blob-dv.puffin",
+		"iceberg-go test fixture", []fixtureBlob{{
+			referencedDataFile: "data/test.parquet",
+			positions:          []uint64{1, 3, 5, 7, 9},
+		}})
+	if err != nil {
+		panic(err)
+	}
+
+	err = writeFixture(outputDir, "multi-blob-dv.puffin",
+		"iceberg-go cross-language fixture", []fixtureBlob{
+			{
+				referencedDataFile: "s3://warehouse/db/table/data/go-file-001.parquet",
+				positions: []uint64{
+					0, 100, 200, (uint64(1) << 32) + 7,
+				},
+			},
+			{
+				referencedDataFile: "s3://warehouse/db/table/data/go-file-002.parquet",
+				positions: []uint64{
+					50, 150, (uint64(2) << 32) + 9,
+				},
+			},
+		})
+	if err != nil {
+		panic(err)
+	}
+
+	position := func(bucket, container, value uint64) uint64 {
+		return (bucket << 32) + (container << 16) + value
+	}
+	allContainerPositions := []uint64{
+		position(0, 0, 5),
+		position(0, 0, 7),
+		position(1, 0, 10),
+		position(1, 0, 20),
+	}
+	for bucket := uint64(0); bucket < 2; bucket++ {
+		for value := uint64(0); value < 10000; value += 2 {
+			allContainerPositions =
+				append(allContainerPositions, position(bucket, 2, value))
+		}
+	}
+	err = writeFixture(outputDir, "all-container-types-dv.puffin",
+		"iceberg-go cross-language fixture", []fixtureBlob{{
+			referencedDataFile: "s3://warehouse/db/table/data/all-containers.parquet",
+			positions:          allContainerPositions,
+			ranges: []positionRange{
+				{start: position(0, 1, 1), end: position(0, 1, 1000)},
+				{start: position(1, 1, 10), end: position(1, 1, 500)},
+			},
+		}})
+	if err != nil {
+		panic(err)
+	}
+}
diff --git a/src/iceberg/test/CMakeLists.txt b/src/iceberg/test/CMakeLists.txt
index 5ca9fd915..f5efed591 100644
--- a/src/iceberg/test/CMakeLists.txt
+++ b/src/iceberg/test/CMakeLists.txt
@@ -158,6 +158,7 @@ add_iceberg_test(util_test
 add_iceberg_test(puffin_test
                  USE_DATA
                  SOURCES
+                 puffin_dv_interop_test.cc
                  puffin_format_test.cc
                  puffin_json_test.cc
                  puffin_reader_writer_test.cc)
diff --git a/src/iceberg/test/puffin_dv_interop_test.cc b/src/iceberg/test/puffin_dv_interop_test.cc
new file mode 100644
index 000000000..b81dcdd51
--- /dev/null
+++ b/src/iceberg/test/puffin_dv_interop_test.cc
@@ -0,0 +1,256 @@
+/*
+ * 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.
+ */
+
+#include <cstddef>
+#include <cstdint>
+#include <filesystem>
+#include <memory>
+#include <span>
+#include <string>
+#include <unordered_set>
+#include <vector>
+
+#include <gtest/gtest.h>
+
+#include "iceberg/deletes/position_delete_index.h"
+#include "iceberg/file_format.h"
+#include "iceberg/manifest/manifest_entry.h"
+#include "iceberg/metadata_columns.h"
+#include "iceberg/puffin/file_metadata.h"
+#include "iceberg/puffin/puffin_reader.h"
+#include "iceberg/test/matchers.h"
+#include "iceberg/test/std_io.h"
+#include "iceberg/test/test_resource.h"
+
+namespace iceberg::puffin {
+
+namespace {
+
+constexpr std::string_view kReferencedDataFileProperty = "referenced-data-file";
+constexpr std::string_view kCardinalityProperty = "cardinality";
+
+int64_t Position(int64_t bucket, int64_t container, int64_t value) {
+  return (bucket << 32) + (container << 16) + value;
+}
+
+std::vector<int64_t> AllContainerPositions() {
+  std::vector<int64_t> positions = {
+      Position(0, 0, 5),
+      Position(0, 0, 7),
+      Position(1, 0, 10),
+      Position(1, 0, 20),
+  };
+  positions.reserve(10004);
+  for (int64_t bucket = 0; bucket < 2; ++bucket) {
+    for (int64_t value = 0; value < 10000; value += 2) {
+      positions.push_back(Position(bucket, 2, value));
+    }
+  }
+  return positions;
+}
+
+struct PositionRange {
+  int64_t begin;
+  int64_t end;
+};
+
+struct ExpectedBlob {
+  std::string referenced_data_file;
+  std::vector<int32_t> input_fields;
+  int64_t offset;
+  int64_t length;
+  int64_t cardinality;
+  std::vector<int64_t> positions;
+  std::vector<PositionRange> ranges;
+};
+
+std::shared_ptr<DataFile> MakeDeleteFile(const ExpectedBlob& expected,
+                                         const std::string& fixture_path,
+                                         int64_t file_size) {
+  return std::make_shared<DataFile>(DataFile{
+      .content = DataFile::Content::kPositionDeletes,
+      .file_path = fixture_path,
+      .file_format = FileFormatType::kPuffin,
+      .record_count = expected.cardinality,
+      .file_size_in_bytes = file_size,
+      .referenced_data_file = expected.referenced_data_file,
+      .content_offset = expected.offset,
+      .content_size_in_bytes = expected.length,
+  });
+}
+
+void AssertPositions(const ExpectedBlob& expected,
+                     const std::shared_ptr<DataFile>& delete_file,
+                     const std::shared_ptr<FileIO>& io) {
+  ASSERT_TRUE(delete_file->content_offset.has_value());
+  ASSERT_TRUE(delete_file->content_size_in_bytes.has_value());
+  ICEBERG_UNWRAP_OR_FAIL(auto input_file, io->NewInputFile(delete_file->file_path));
+  ICEBERG_UNWRAP_OR_FAIL(auto stream, input_file->Open());
+  std::vector<std::byte> data(static_cast<size_t>(*delete_file->content_size_in_bytes));
+  ASSERT_THAT(stream->ReadFully(*delete_file->content_offset, data), IsOk());
+
+  std::span<const uint8_t> blob(reinterpret_cast<const uint8_t*>(data.data()),
+                                data.size());
+  ICEBERG_UNWRAP_OR_FAIL(auto positions,
+                         PositionDeleteIndex::Deserialize(blob, delete_file));
+  EXPECT_EQ(positions.Cardinality(), expected.cardinality);
+  int64_t expected_cardinality = static_cast<int64_t>(expected.positions.size());
+  for (const auto& range : expected.ranges) {
+    expected_cardinality += range.end - range.begin;
+  }
+  ASSERT_EQ(expected_cardinality, expected.cardinality);
+  for (int64_t position : expected.positions) {
+    EXPECT_TRUE(positions.IsDeleted(position)) << "Missing position " << position;
+  }
+  for (const auto& range : expected.ranges) {
+    for (int64_t position = range.begin; position < range.end; ++position) {
+      ASSERT_TRUE(positions.IsDeleted(position)) << "Missing position " << position;
+    }
+  }
+}
+
+void AssertFixture(const std::string& resource_name,
+                   const std::vector<ExpectedBlob>& expected_blobs) {
+  const std::string fixture_path = GetResourcePath(resource_name);
+  const auto file_size = static_cast<int64_t>(std::filesystem::file_size(fixture_path));
+  auto io = std::make_shared<test::StdFileIO>();
+  ICEBERG_UNWRAP_OR_FAIL(auto input_file, io->NewInputFile(fixture_path));
+  ICEBERG_UNWRAP_OR_FAIL(auto reader, PuffinReader::Make(std::move(input_file)));
+  ICEBERG_UNWRAP_OR_FAIL(auto metadata, reader->ReadFileMetadata());
+
+  ASSERT_EQ(metadata.blobs.size(), expected_blobs.size());
+  std::unordered_set<std::string> seen;
+  for (const auto& blob : metadata.blobs) {
+    const auto& referenced_data_file =
+        blob.properties.at(std::string(kReferencedDataFileProperty));
+    auto expected = std::ranges::find_if(expected_blobs, [&](const auto& candidate) {
+      return candidate.referenced_data_file == referenced_data_file;
+    });
+    ASSERT_NE(expected, expected_blobs.end())
+        << "Unexpected referenced data file " << referenced_data_file;
+    ASSERT_TRUE(seen.insert(referenced_data_file).second)
+        << "Duplicate referenced data file " << referenced_data_file;
+
+    EXPECT_EQ(blob.type, StandardBlobTypes::kDeletionVectorV1);
+    EXPECT_EQ(blob.input_fields, expected->input_fields);
+    EXPECT_EQ(blob.snapshot_id, -1);
+    EXPECT_EQ(blob.sequence_number, -1);
+    EXPECT_EQ(blob.offset, expected->offset);
+    EXPECT_EQ(blob.length, expected->length);
+    EXPECT_TRUE(blob.compression_codec.empty());
+    EXPECT_EQ(blob.properties.at(std::string(kCardinalityProperty)),
+              std::to_string(expected->cardinality));
+
+    AssertPositions(*expected, MakeDeleteFile(*expected, fixture_path, file_size), io);
+  }
+  EXPECT_EQ(seen.size(), expected_blobs.size());
+}
+
+}  // namespace
+
+TEST(PuffinDVInteropTest, ReadsJavaSingleBlobFixture) {
+  AssertFixture(
+      "deletion_vectors/java/single-blob-dv.puffin",
+      {{
+          .referenced_data_file = "s3://warehouse/db/table/data/00000-0-abc.parquet",
+          .input_fields = {MetadataColumns::kFilePositionColumnId},
+          .offset = 4,
+          .length = 50,
+          .cardinality = 5,
+          .positions = {1, 3, 5, 7, 9},
+      }});
+}
+
+TEST(PuffinDVInteropTest, ReadsJavaMultiBlobFixture) {
+  AssertFixture(
+      "deletion_vectors/java/multi-blob-dv.puffin",
+      {
+          {
+              .referenced_data_file = "s3://warehouse/db/table/data/file-001.parquet",
+              .input_fields = {MetadataColumns::kFilePositionColumnId},
+              .offset = 4,
+              .length = 46,
+              .cardinality = 3,
+              .positions = {0, 100, 200},
+          },
+          {
+              .referenced_data_file = "s3://warehouse/db/table/data/file-002.parquet",
+              .input_fields = {MetadataColumns::kFilePositionColumnId},
+              .offset = 50,
+              .length = 44,
+              .cardinality = 2,
+              .positions = {50, 150},
+          },
+      });
+}
+
+TEST(PuffinDVInteropTest, ReadsGoSingleBlobFixture) {
+  AssertFixture("deletion_vectors/go/single-blob-dv.puffin",
+                {{
+                    .referenced_data_file = "data/test.parquet",
+                    .input_fields = {},
+                    .offset = 4,
+                    .length = 50,
+                    .cardinality = 5,
+                    .positions = {1, 3, 5, 7, 9},
+                }});
+}
+
+TEST(PuffinDVInteropTest, ReadsGoMultiBlobFixture) {
+  AssertFixture(
+      "deletion_vectors/go/multi-blob-dv.puffin",
+      {
+          {
+              .referenced_data_file = "s3://warehouse/db/table/data/go-file-001.parquet",
+              .input_fields = {},
+              .offset = 4,
+              .length = 68,
+              .cardinality = 4,
+              .positions = {0, 100, 200, (int64_t{1} << 32) + 7},
+          },
+          {
+              .referenced_data_file = "s3://warehouse/db/table/data/go-file-002.parquet",
+              .input_fields = {},
+              .offset = 72,
+              .length = 66,
+              .cardinality = 3,
+              .positions = {50, 150, (int64_t{2} << 32) + 9},
+          },
+      });
+}
+
+TEST(PuffinDVInteropTest, ReadsGoAllContainerTypesFixture) {
+  AssertFixture(
+      "deletion_vectors/go/all-container-types-dv.puffin",
+      {{
+          .referenced_data_file = "s3://warehouse/db/table/data/all-containers.parquet",
+          .input_fields = {},
+          .offset = 4,
+          .length = 16466,
+          .cardinality = 11493,
+          .positions = AllContainerPositions(),
+          .ranges =
+              {
+                  {Position(0, 1, 1), Position(0, 1, 1000)},
+                  {Position(1, 1, 10), Position(1, 1, 500)},
+              },
+      }});
+}
+
+}  // namespace iceberg::puffin
diff --git a/src/iceberg/test/resources/deletion_vectors/README.md b/src/iceberg/test/resources/deletion_vectors/README.md
new file mode 100644
index 000000000..983cbfe3e
--- /dev/null
+++ b/src/iceberg/test/resources/deletion_vectors/README.md
@@ -0,0 +1,44 @@
+# Deletion vector interoperability fixtures
+
+These fixtures verify deletion-vector compatibility against bytes produced by
+implementations other than Iceberg C++.
+
+## Java fixtures
+
+The files under `java/` were copied byte-for-byte from Apache Iceberg Go commit
+`3020adbbc3faff047da6f483f739f1b5e1de611b`, where they are consumed by
+`table/dv/dv_cross_client_test.go`.
+
+- `single-blob-dv.puffin` and `multi-blob-dv.puffin` were produced by Apache
+  Iceberg Java using
+  `iceberg-go/dev/dv-fixtures/GenerateDVFixtures.java`.
+
+The committed Puffin bytes were reproduced against Apache Iceberg Java commit
+`76d35b1e40f77edcad19646bb6afdd9f05249964`. To regenerate them, follow
+`iceberg-go/dev/dv-fixtures/README.md` at the Iceberg Go commit above using
+that Java revision, then copy the generated files into `java/`.
+
+## Go fixtures
+
+The files under `go/` are produced with the Iceberg Go Puffin and deletion
+vector writers. From an Iceberg Go checkout at commit
+`3020adbbc3faff047da6f483f739f1b5e1de611b`, run:
+
+```bash
+go run /path/to/iceberg-cpp/dev/dv-fixtures/generate_go_fixtures.go \
+  /path/to/iceberg-cpp/src/iceberg/test/resources/deletion_vectors/go
+```
+
+The multi-blob fixture includes positions in three distinct high-32-bit
+buckets. `all-container-types-dv.puffin` covers array, run, and bitmap
+containers inside a complete Puffin file.
+
+## Checksums
+
+```text
+ab8309671c0c5ef1956f4f1d7b907f4a69ffeb154ab77d5a4855d7dd4779108c  java/single-blob-dv.puffin
+fed7edbb5a343a6c6fc4706ff4c213ab3f0a50916baeb228619f1f2c956f3f27  java/multi-blob-dv.puffin
+dd293e827439cd22053a9356ae6d56f0d5d69e648b1850d39663cdc5c7ec5a77  go/single-blob-dv.puffin
+000a77697d5ddce01e0242786b91fc15482a8b8f4fa3c8e214a936d7c2fc978b  go/multi-blob-dv.puffin
+d9f374706891b4780f8a59f155a7cc4cae4161b6415578abb4790f6eed793a52  go/all-container-types-dv.puffin
+```
diff --git a/src/iceberg/test/resources/deletion_vectors/go/all-container-types-dv.puffin b/src/iceberg/test/resources/deletion_vectors/go/all-container-types-dv.puffin
new file mode 100644
index 0000000000000000000000000000000000000000..5582a008e1adcba5868684e7169fffc95a351171
GIT binary patch
literal 16781
zcmeI)u};G<5C-712qCe;1ArHnnh;S{A}lb#z{JuCF*wO3wPfsIJ3v)+=8<>~#2YXm
z^)0vvq#_=n`kf`Vth@6!c*!ZZ*Oyt6T->~WoK?eQJ<d;t3C{=od~FSr*S+mz=XW;6
zfB*y_u(1N`YY=Y`fB*y_009U<00Izz00bZa0SG_<0uX=z1Rwwb2tWV=5ZH8q0Uz*B
zd%$YYw+HMd``jL|o2&*K`&AGJg8&2|Fx~VE9_NDq1Rwwb2tWV=5P$##AOHafKmY;|
zfB*y_009U<00Izzz*^w*>h18B|0=K$g>gj?`Q1Xqc}pS}6&Z@!S@}d|bY4DXVkBx!
zrs}udk3_Jl4Ko*|u0$>~UZBU0Y)R5~4+Z(3MOW8>qR7W1VaPUdCh}}Do=irf^{yo!
zHF3O!D3!0YRYu1-C(p9+*(vAqRFm&xR<cr2$y%Gfws4xK>9g`Qb6udcD$=M5!!Nui
zRmRBD*{HN7AC6n)xvVIbORo7}PERUw=%P=lOX8v%*|>w!yAY&Nw&_$u2ep2QohOcP
L)VEXc=*hkTRU)zo

literal 0
HcmV?d00001

diff --git a/src/iceberg/test/resources/deletion_vectors/go/multi-blob-dv.puffin b/src/iceberg/test/resources/deletion_vectors/go/multi-blob-dv.puffin
new file mode 100644
index 0000000000000000000000000000000000000000..4404137af3286dfc595d7dc47568d7d06cf902af
GIT binary patch
literal 650
zcmcIiJx{|h5OupCHhw~N6GKdzwiK{fSeUyb1}DCxmW&;2r%+Y>8_fI_Hde&I-(Y6q
zk_u5(f(7Bl_wn@Z-m_<y7a3#h{N?pb^(pu|V#rvpOZx1fyA^xxD&HN}b@tfD)8iXK
z#LpbP4mn~EtevsTu+K7ccs!#`R*)N)2M|{)h|3y5fI@>&JIfbXMCbV;g9J*AMzyf(
z1cH@yn7hce0>LvX;BAApKyI6x9Q~%~$}(UCF-gFnt>PTS^fZCmyBdAeNMtK0q*vNX
zqvMh$PiT<G5`9aixRO!wQk!<EFcxWgCq2$x6EIbI8f9)MAWT}h4#tPW?5LKWmWi-l
zC;z4BWV9t|`n#m@pCyg9B$4?${&|#<tjb&abb;i4#Va=`ybFOFX{$z7I4JdfY&?=F
L@0%~i+PD7%?uWWS

literal 0
HcmV?d00001

diff --git a/src/iceberg/test/resources/deletion_vectors/go/single-blob-dv.puffin b/src/iceberg/test/resources/deletion_vectors/go/single-blob-dv.puffin
new file mode 100644
index 0000000000000000000000000000000000000000..84c6baf796fc5cd89a2b94f417f0aaf0d4c3354f
GIT binary patch
literal 314
zcmZ8dy-vh14E6~SV&f4ikEnsG#0_L)VZI$Pq>a;y6q1s<Cr;Hp6+7>McojAl91fQA
zWPkGaW3M0X8vwZa_`I_VY5q$u0S@yiTne~=6F7r&xPZ6s>U}M&8QYNB5%uGYyc-BL
zS_;HIMYW}lQ&!ssYjmE1Eo)CTj?oO`Cb9AswQ2;Qmx*FWDo)QW<pY?U8;Mch))*-E
zyg|KM);Q!eP-ageHe+Y9_0a^+J8@o#Q6@*Zh*@O~8|8fX4JG{z)5zC@$>PSi-%FGu
gwM>SE*cLJ0k!0yrpQ^&EoPXshlkmR&-c#=X0SLfd%>V!Z

literal 0
HcmV?d00001

diff --git a/src/iceberg/test/resources/deletion_vectors/java/multi-blob-dv.puffin b/src/iceberg/test/resources/deletion_vectors/java/multi-blob-dv.puffin
new file mode 100644
index 0000000000000000000000000000000000000000..b9037900c2cab7f0d8b869f477a81440816a5df8
GIT binary patch
literal 639
zcmcJNy-ve05Xap<0lYzE0Rkx|A59BnLqbSwomdcqV_y<WCJwe!sH(0!0uwL7!pw?z
z2_Az>3la?7z|-C7e;+^JP0vpgLdf3R`-vQ&@%{3ckhMP~TdSZXujHco*hOb&vwE;n
zr{s}5JiR<mvHF4*+7yoFR}1RrEzq1w(BPG6*bIa>mdz3x(o%tzy(~?#qwIJz$;Q_q
zbq#M_Z9G$w<}ASg+;-3iU`=;ZfL#YoSvv4E&qAu9seFy*B&4l1Em*Gr%Py#e5^SF;
zndF|cQt5seH_D^vj$5cr=OB_r<awddKuy$ZSR5yZEw|X$<D7z9sT!`8pJVojE|=l|
zgN@@KWJvjEWa&n-6v=P_zdH#F+~WrpJ-ZTs8>m=ig0ss&SmPWMT0!H3Qr-K`f(!O}
T%S8>*m-YaQZV%T-)*r_w%Cx|g

literal 0
HcmV?d00001

diff --git a/src/iceberg/test/resources/deletion_vectors/java/single-blob-dv.puffin b/src/iceberg/test/resources/deletion_vectors/java/single-blob-dv.puffin
new file mode 100644
index 0000000000000000000000000000000000000000..5c0007e217711b51d5ee7be20b589f4d0aa0a39c
GIT binary patch
literal 384
zcmY*V%TB{E5Of6yapM!BJpqbILQ@{TAtBB^aX}m$d(&7lcCej7RrRa5^ACv6>4lA2
z4(w@nG&?hz`Qv@X7`yoXxziI`{zjCH?e0T%O5lVYup@TNPT1!*`<m0&8XD`H2;zAS
zY1JbL&}cCk@Awi`@_{ca$e}f8^+>y_>YMs@dR@<6a)?g$vGa)=4T4t$;M;&sA$P-T
zgJB2ywvCuT)Hzsmi`0Rbl{xgm_ZX5vl3hc<7Q^_ext2-t*4QyjoQk4&mjOFJL@abu
zB-vP6h*Ol~@RG|$UG_4Nb)u+B25p?QCap+d2Ad5f{colM%0y1h$S?{i>VhwP_VARc
h;A7;<8gwaZ%|{vn#_UY?QgvADx#uJ*_C=2^{sKt2a)JN=

literal 0
HcmV?d00001


From dc84d6e3819c026ed9f5b998edcbcd40493fdd8c Mon Sep 17 00:00:00 2001
From: xuanyili <xuanyili@uber.com>
Date: Sat, 5 Sep 2026 18:53:36 +0000
Subject: [PATCH 2/4] fix: enforce deletion vector scan applicability

---
 src/iceberg/delete_file_index.cc           |  19 +++-
 src/iceberg/test/delete_file_index_test.cc | 102 +++++++++++++++++++--
 2 files changed, 109 insertions(+), 12 deletions(-)

diff --git a/src/iceberg/delete_file_index.cc b/src/iceberg/delete_file_index.cc
index a8c4ef126..7a59c2b26 100644
--- a/src/iceberg/delete_file_index.cc
+++ b/src/iceberg/delete_file_index.cc
@@ -41,6 +41,7 @@
 #include "iceberg/util/content_file_util.h"
 #include "iceberg/util/executor_util_internal.h"
 #include "iceberg/util/macros.h"
+#include "iceberg/util/struct_like_set.h"
 
 namespace iceberg {
 
@@ -453,10 +454,20 @@ Result<std::shared_ptr<DataFile>> DeleteFileIndex::FindDV(
     return nullptr;
   }
 
-  ICEBERG_CHECK(it->second.sequence_number.value() >= seq,
-                "DV data sequence number {} must be greater than or equal to data file "
-                "sequence number {}",
-                it->second.sequence_number.value(), seq);
+  const auto& dv = *it->second.data_file;
+  ICEBERG_PRECHECK(data_file.partition_spec_id.has_value(),
+                   "Missing partition spec id from data file {}", data_file.file_path);
+  ICEBERG_PRECHECK(dv.partition_spec_id.has_value(),
+                   "Missing partition spec id from DV {}", dv.file_path);
+  if (dv.partition_spec_id != data_file.partition_spec_id) {
+    return nullptr;
+  }
+
+  ICEBERG_ASSIGN_OR_RAISE(auto partitions_match,
+                          StructLikeEqual(dv.partition, data_file.partition));
+  if (!partitions_match || it->second.sequence_number.value() < seq) {
+    return nullptr;
+  }
 
   return it->second.data_file;
 }
diff --git a/src/iceberg/test/delete_file_index_test.cc b/src/iceberg/test/delete_file_index_test.cc
index 75c82bc39..340688632 100644
--- a/src/iceberg/test/delete_file_index_test.cc
+++ b/src/iceberg/test/delete_file_index_test.cc
@@ -65,6 +65,11 @@ class DeleteFileIndexTest : public testing::TestWithParam<int8_t> {
         PartitionSpec::Make(
             /*spec_id=*/1, {PartitionField(/*source_id=*/2, /*field_id=*/1000,
                                            "data_bucket", Transform::Bucket(16))}));
+    ICEBERG_UNWRAP_OR_FAIL(
+        equivalent_partitioned_spec_,
+        PartitionSpec::Make(
+            /*spec_id=*/2, {PartitionField(/*source_id=*/2, /*field_id=*/1000,
+                                           "data_bucket", Transform::Bucket(16))}));
 
     // Unpartitioned spec
     unpartitioned_spec_ = PartitionSpec::Unpartitioned();
@@ -187,6 +192,7 @@ class DeleteFileIndexTest : public testing::TestWithParam<int8_t> {
 
   std::unordered_map<int32_t, std::shared_ptr<PartitionSpec>> GetSpecsById() {
     return {{partitioned_spec_->spec_id(), partitioned_spec_},
+            {equivalent_partitioned_spec_->spec_id(), equivalent_partitioned_spec_},
             {unpartitioned_spec_->spec_id(), unpartitioned_spec_}};
   }
 
@@ -209,6 +215,7 @@ class DeleteFileIndexTest : public testing::TestWithParam<int8_t> {
   std::shared_ptr<FileIO> file_io_;
   std::shared_ptr<Schema> schema_;
   std::shared_ptr<PartitionSpec> partitioned_spec_;
+  std::shared_ptr<PartitionSpec> equivalent_partitioned_spec_;
   std::shared_ptr<PartitionSpec> unpartitioned_spec_;
 
   std::shared_ptr<DataFile> file_a_;
@@ -1041,8 +1048,9 @@ TEST_P(DeleteFileIndexTest, TestMixDeleteFilesAndDVs) {
   auto partition_b = PartitionValues({Literal::Int(1)});
 
   // Position delete for file_a_
-  auto pos_delete_a = MakePositionDeleteFile("/path/to/pos-delete-a.parquet", partition_a,
-                                             partitioned_spec_->spec_id());
+  auto pos_delete_a =
+      MakePositionDeleteFile("/path/to/pos-delete-a.parquet", partition_a,
+                             partitioned_spec_->spec_id(), file_a_->file_path);
   // DV for file_a_ (should take precedence)
   auto dv_a = MakeDV("/path/to/dv-a.puffin", partition_a, partitioned_spec_->spec_id(),
                      file_a_->file_path);
@@ -1113,7 +1121,82 @@ TEST_P(DeleteFileIndexTest, TestMultipleDVs) {
   EXPECT_THAT(index_result, HasErrorMessage(file_a_->file_path));
 }
 
-TEST_P(DeleteFileIndexTest, TestInvalidDVSequenceNumber) {
+TEST_P(DeleteFileIndexTest, TestDVApplicability) {
+  auto version = GetParam();
+  if (version < 3) {
+    GTEST_SKIP() << "DVs only supported in V3+";
+  }
+
+  const auto null_partition = PartitionValues({Literal::Null(int32())});
+  auto null_partition_file = MakeDataFile("/path/to/data-null.parquet", null_partition,
+                                          partitioned_spec_->spec_id());
+
+  struct TestCase {
+    std::string name;
+    PartitionValues dv_partition;
+    std::shared_ptr<PartitionSpec> dv_spec;
+    std::shared_ptr<DataFile> data_file;
+    bool applies;
+  };
+  const std::vector<TestCase> cases = {
+      {
+          .name = "equal-partition",
+          .dv_partition = file_a_->partition,
+          .dv_spec = partitioned_spec_,
+          .data_file = file_a_,
+          .applies = true,
+      },
+      {
+          .name = "different-spec",
+          .dv_partition = file_a_->partition,
+          .dv_spec = equivalent_partitioned_spec_,
+          .data_file = file_a_,
+          .applies = false,
+      },
+      {
+          .name = "different-partition-value",
+          .dv_partition = file_b_->partition,
+          .dv_spec = partitioned_spec_,
+          .data_file = file_a_,
+          .applies = false,
+      },
+      {
+          .name = "equal-null-partition",
+          .dv_partition = null_partition,
+          .dv_spec = partitioned_spec_,
+          .data_file = null_partition_file,
+          .applies = true,
+      },
+      {
+          .name = "null-partition-mismatch",
+          .dv_partition = file_a_->partition,
+          .dv_spec = partitioned_spec_,
+          .data_file = null_partition_file,
+          .applies = false,
+      },
+  };
+
+  for (const auto& test_case : cases) {
+    SCOPED_TRACE(test_case.name);
+    auto dv = MakeDV("/path/to/" + test_case.name + ".puffin", test_case.dv_partition,
+                     test_case.dv_spec->spec_id(), test_case.data_file->file_path);
+    std::vector<ManifestEntry> entries;
+    entries.push_back(MakeDeleteEntry(/*snapshot_id=*/1000L, /*sequence_number=*/2, dv));
+    auto manifest = WriteDeleteManifest(version, /*snapshot_id=*/1000L,
+                                        std::move(entries), test_case.dv_spec);
+    ICEBERG_UNWRAP_OR_FAIL(auto index, BuildIndex({manifest}));
+    ICEBERG_UNWRAP_OR_FAIL(auto deletes, index->ForDataFile(1, *test_case.data_file));
+
+    if (test_case.applies) {
+      ASSERT_EQ(deletes.size(), 1);
+      EXPECT_EQ(deletes[0]->file_path, dv->file_path);
+    } else {
+      EXPECT_TRUE(deletes.empty());
+    }
+  }
+}
+
+TEST_P(DeleteFileIndexTest, TestInapplicableDVSequenceNumber) {
   auto version = GetParam();
   if (version < 3) {
     GTEST_SKIP() << "DVs only supported in V3+";
@@ -1123,20 +1206,23 @@ TEST_P(DeleteFileIndexTest, TestInvalidDVSequenceNumber) {
 
   auto dv = MakeDV("/path/to/dv.puffin", partition_a, partitioned_spec_->spec_id(),
                    file_a_->file_path);
+  auto pos_delete =
+      MakePositionDeleteFile("/path/to/pos-delete.parquet", partition_a,
+                             partitioned_spec_->spec_id(), file_a_->file_path);
 
   std::vector<ManifestEntry> entries;
   entries.push_back(MakeDeleteEntry(/*snapshot_id=*/1000L, /*sequence_number=*/1, dv));
+  entries.push_back(
+      MakeDeleteEntry(/*snapshot_id=*/1000L, /*sequence_number=*/2, pos_delete));
 
   auto manifest = WriteDeleteManifest(version, /*snapshot_id=*/1000L, std::move(entries),
                                       partitioned_spec_);
 
   ICEBERG_UNWRAP_OR_FAIL(auto index, BuildIndex({manifest}));
 
-  // Querying with sequence number > DV sequence number should fail
-  auto result = index->ForDataFile(2, *file_a_);
-  EXPECT_THAT(result, IsError(ErrorKind::kValidationFailed));
-  EXPECT_THAT(result, HasErrorMessage(
-                          "must be greater than or equal to data file sequence number"));
+  ICEBERG_UNWRAP_OR_FAIL(auto deletes, index->ForDataFile(2, *file_a_));
+  ASSERT_EQ(deletes.size(), 1);
+  EXPECT_EQ(deletes[0]->file_path, pos_delete->file_path);
 }
 
 TEST_P(DeleteFileIndexTest, TestReferencedDeleteFiles) {

From 651567939eb69f6ea425798249156a84fd9618ca Mon Sep 17 00:00:00 2001
From: xuanyili <xuanyili@uber.com>
Date: Sat, 5 Sep 2026 18:24:25 +0000
Subject: [PATCH 3/4] fix: validate deletion vectors against Puffin metadata

---
 src/iceberg/deletes/dv_util.cc                |  83 ++++-
 .../deletes/roaring_position_bitmap.cc        |   1 +
 src/iceberg/test/CMakeLists.txt               |   1 +
 src/iceberg/test/dv_util_test.cc              | 303 ++++++++++++++++++
 src/iceberg/test/puffin_dv_interop_test.cc    |  13 +-
 5 files changed, 380 insertions(+), 21 deletions(-)
 create mode 100644 src/iceberg/test/dv_util_test.cc

diff --git a/src/iceberg/deletes/dv_util.cc b/src/iceberg/deletes/dv_util.cc
index 670ccfedb..e9fb6449d 100644
--- a/src/iceberg/deletes/dv_util.cc
+++ b/src/iceberg/deletes/dv_util.cc
@@ -17,6 +17,7 @@
  * under the License.
  */
 
+#include <algorithm>
 #include <cstddef>
 #include <cstdint>
 #include <format>
@@ -36,13 +37,63 @@
 #include "iceberg/metadata_columns.h"
 #include "iceberg/partition_spec.h"
 #include "iceberg/puffin/file_metadata.h"
+#include "iceberg/puffin/puffin_reader.h"
 #include "iceberg/result.h"
 #include "iceberg/util/content_file_util.h"
 #include "iceberg/util/macros.h"
+#include "iceberg/util/string_util.h"
 #include "iceberg/version.h"
 
 namespace iceberg {
 
+namespace {
+
+constexpr std::string_view kReferencedDataFileProperty = "referenced-data-file";
+constexpr std::string_view kCardinalityProperty = "cardinality";
+
+Status ValidateDVBlobMetadata(const puffin::BlobMetadata& blob,
+                              const DataFile& delete_file) {
+  ICEBERG_PRECHECK(blob.type == puffin::StandardBlobTypes::kDeletionVectorV1,
+                   "Invalid deletion vector blob type '{}', expected '{}'", blob.type,
+                   puffin::StandardBlobTypes::kDeletionVectorV1);
+  ICEBERG_PRECHECK(blob.snapshot_id == -1,
+                   "Deletion vector requires snapshot-id -1, got {}", blob.snapshot_id);
+  ICEBERG_PRECHECK(blob.sequence_number == -1,
+                   "Deletion vector requires sequence-number -1, got {}",
+                   blob.sequence_number);
+  ICEBERG_PRECHECK(blob.compression_codec.empty(),
+                   "Deletion vector must not be compressed, got '{}'",
+                   blob.compression_codec);
+
+  auto referenced_data_file =
+      blob.properties.find(std::string(kReferencedDataFileProperty));
+  ICEBERG_PRECHECK(referenced_data_file != blob.properties.end() &&
+                       !referenced_data_file->second.empty(),
+                   "Deletion vector blob requires non-empty '{}' property",
+                   kReferencedDataFileProperty);
+  ICEBERG_PRECHECK(
+      referenced_data_file->second == *delete_file.referenced_data_file,
+      "Manifest referenced_data_file '{}' does not match Puffin '{}' property '{}'",
+      *delete_file.referenced_data_file, kReferencedDataFileProperty,
+      referenced_data_file->second);
+
+  auto cardinality = blob.properties.find(std::string(kCardinalityProperty));
+  ICEBERG_PRECHECK(cardinality != blob.properties.end(),
+                   "Deletion vector blob requires '{}' property", kCardinalityProperty);
+  ICEBERG_ASSIGN_OR_RAISE(auto parsed_cardinality,
+                          StringUtils::ParseNumber<int64_t>(cardinality->second));
+  ICEBERG_PRECHECK(parsed_cardinality >= 0,
+                   "Deletion vector cardinality must be non-negative, got {}",
+                   parsed_cardinality);
+  ICEBERG_PRECHECK(
+      parsed_cardinality == delete_file.record_count,
+      "Manifest record_count {} does not match Puffin cardinality property {}",
+      delete_file.record_count, parsed_cardinality);
+  return {};
+}
+
+}  // namespace
+
 Result<std::vector<std::shared_ptr<DataFile>>> DVUtil::MergeAndWriteDVs(
     std::span<const DeletionVectorMergeGroup> groups, std::string_view output_path,
     const std::shared_ptr<FileIO>& io) {
@@ -84,6 +135,10 @@ Result<PositionDeleteIndex> DVUtil::ReadDV(const std::shared_ptr<DataFile>& dele
           delete_file->content_size_in_bytes.has_value(),
       "Deletion vector requires content_offset and content_size_in_bytes: {}",
       delete_file->file_path);
+  ICEBERG_PRECHECK(delete_file->referenced_data_file.has_value() &&
+                       !delete_file->referenced_data_file->empty(),
+                   "Deletion vector requires referenced_data_file: {}",
+                   delete_file->file_path);
 
   const int64_t offset = delete_file->content_offset.value();
   const int64_t length = delete_file->content_size_in_bytes.value();
@@ -94,14 +149,21 @@ Result<PositionDeleteIndex> DVUtil::ReadDV(const std::shared_ptr<DataFile>& dele
                    "Cannot read deletion vector larger than 2GB: {}", length);
 
   ICEBERG_ASSIGN_OR_RAISE(auto input_file, io->NewInputFile(delete_file->file_path));
-  ICEBERG_ASSIGN_OR_RAISE(auto stream, input_file->Open());
-
-  std::vector<std::byte> bytes(static_cast<size_t>(length));
-  ICEBERG_RETURN_UNEXPECTED(stream->ReadFully(offset, bytes));
-  ICEBERG_RETURN_UNEXPECTED(stream->Close());
-
-  std::span<const uint8_t> blob(reinterpret_cast<const uint8_t*>(bytes.data()),
-                                bytes.size());
+  ICEBERG_ASSIGN_OR_RAISE(auto reader, puffin::PuffinReader::Make(std::move(input_file)));
+  ICEBERG_ASSIGN_OR_RAISE(auto metadata, reader->ReadFileMetadata());
+  auto blob_metadata = std::ranges::find_if(
+      metadata.blobs, [offset](const auto& blob) { return blob.offset == offset; });
+  ICEBERG_PRECHECK(blob_metadata != metadata.blobs.end(),
+                   "No Puffin blob starts at manifest content_offset {}", offset);
+  ICEBERG_PRECHECK(
+      blob_metadata->length == length,
+      "Puffin blob at offset {} has length {}, manifest content_size_in_bytes is {}",
+      offset, blob_metadata->length, length);
+  ICEBERG_RETURN_UNEXPECTED(ValidateDVBlobMetadata(*blob_metadata, *delete_file));
+
+  ICEBERG_ASSIGN_OR_RAISE(auto blob_data, reader->ReadBlob(*blob_metadata));
+  std::span<const uint8_t> blob(reinterpret_cast<const uint8_t*>(blob_data.second.data()),
+                                blob_data.second.size());
   return PositionDeleteIndex::Deserialize(blob, delete_file);
 }
 
@@ -118,8 +180,9 @@ Result<puffin::BlobMetadata> DVUtil::WriteDVBlob(puffin::PuffinWriter& writer,
       .data = std::move(data),
       .requested_compression = puffin::PuffinCompressionCodec::kNone,
   };
-  blob.properties.emplace("referenced-data-file", std::string(referenced_data_file));
-  blob.properties.emplace("cardinality", std::format("{}", positions.Cardinality()));
+  blob.properties.emplace(kReferencedDataFileProperty, std::string(referenced_data_file));
+  blob.properties.emplace(kCardinalityProperty,
+                          std::format("{}", positions.Cardinality()));
   return writer.Write(blob);
 }
 
diff --git a/src/iceberg/deletes/roaring_position_bitmap.cc b/src/iceberg/deletes/roaring_position_bitmap.cc
index a2827d4bb..2020160f1 100644
--- a/src/iceberg/deletes/roaring_position_bitmap.cc
+++ b/src/iceberg/deletes/roaring_position_bitmap.cc
@@ -271,6 +271,7 @@ Result<RoaringPositionBitmap> RoaringPositionBitmap::Deserialize(std::string_vie
     --remaining_count;
   }
 
+  ICEBERG_PRECHECK(remaining == 0, "Trailing data after bitmaps: {} bytes", remaining);
   return RoaringPositionBitmap(std::move(impl));
 }
 
diff --git a/src/iceberg/test/CMakeLists.txt b/src/iceberg/test/CMakeLists.txt
index f5efed591..3238e9c2d 100644
--- a/src/iceberg/test/CMakeLists.txt
+++ b/src/iceberg/test/CMakeLists.txt
@@ -158,6 +158,7 @@ add_iceberg_test(util_test
 add_iceberg_test(puffin_test
                  USE_DATA
                  SOURCES
+                 dv_util_test.cc
                  puffin_dv_interop_test.cc
                  puffin_format_test.cc
                  puffin_json_test.cc
diff --git a/src/iceberg/test/dv_util_test.cc b/src/iceberg/test/dv_util_test.cc
new file mode 100644
index 000000000..d84eb0784
--- /dev/null
+++ b/src/iceberg/test/dv_util_test.cc
@@ -0,0 +1,303 @@
+/*
+ * 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.
+ */
+
+#include <zconf.h>
+#include <zlib.h>
+
+#include <array>
+#include <cstddef>
+#include <cstdint>
+#include <cstring>
+#include <functional>
+#include <memory>
+#include <span>
+#include <string>
+#include <string_view>
+#include <utility>
+#include <vector>
+
+#include <gtest/gtest.h>
+
+#include "iceberg/deletes/dv_util_internal.h"
+#include "iceberg/deletes/position_delete_index.h"
+#include "iceberg/file_format.h"
+#include "iceberg/manifest/manifest_entry.h"
+#include "iceberg/metadata_columns.h"
+#include "iceberg/puffin/file_metadata.h"
+#include "iceberg/puffin/json_serde_internal.h"
+#include "iceberg/puffin/puffin_format.h"
+#include "iceberg/test/matchers.h"
+#include "iceberg/test/mock_io.h"
+#include "iceberg/util/endian.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg {
+
+namespace {
+
+constexpr std::string_view kReferencedDataFile = "data.parquet";
+constexpr std::string_view kReferencedDataFileProperty = "referenced-data-file";
+constexpr std::string_view kCardinalityProperty = "cardinality";
+
+struct MetadataTestCase {
+  std::string name;
+  std::function<void(std::vector<uint8_t>&)> mutate_data;
+  std::function<void(puffin::Blob&)> mutate_blob;
+  std::function<void(DataFile&)> mutate_manifest;
+  std::string expected_error;
+};
+
+void AddTrailingBitmapData(std::vector<uint8_t>& data) {
+  constexpr size_t kLengthPrefixBytes = 4;
+  constexpr size_t kCrcBytes = 4;
+
+  data.insert(data.end() - kCrcBytes, 0);
+  const auto length = static_cast<int32_t>(data.size() - kLengthPrefixBytes - kCrcBytes);
+  WriteBigEndian(length, data.data());
+
+  uLong crc = crc32(0L, Z_NULL, 0);
+  crc = crc32(crc, reinterpret_cast<const Bytef*>(data.data() + kLengthPrefixBytes),
+              static_cast<uInt>(length));
+  WriteBigEndian(static_cast<uint32_t>(crc), data.data() + data.size() - kCrcBytes);
+}
+
+Result<std::shared_ptr<DataFile>> WriteDVFixture(const std::shared_ptr<MockFileIO>& io,
+                                                 const MetadataTestCase& test_case) {
+  PositionDeleteIndex positions;
+  positions.Delete(1);
+  positions.Delete(3);
+  positions.Delete(5);
+  ICEBERG_ASSIGN_OR_RAISE(auto data, positions.Serialize());
+  if (test_case.mutate_data) {
+    test_case.mutate_data(data);
+  }
+
+  puffin::Blob blob{
+      .type = std::string(puffin::StandardBlobTypes::kDeletionVectorV1),
+      .input_fields = {MetadataColumns::kFilePositionColumnId},
+      .snapshot_id = -1,
+      .sequence_number = -1,
+      .data = std::move(data),
+      .requested_compression = puffin::PuffinCompressionCodec::kNone,
+      .properties =
+          {
+              {std::string(kReferencedDataFileProperty),
+               std::string(kReferencedDataFile)},
+              {std::string(kCardinalityProperty), "3"},
+          },
+  };
+  if (test_case.mutate_blob) {
+    test_case.mutate_blob(blob);
+  }
+
+  const std::string path = "memory://" + test_case.name + ".puffin";
+  const auto codec =
+      blob.requested_compression.value_or(puffin::PuffinCompressionCodec::kNone);
+  puffin::BlobMetadata blob_metadata{
+      .type = blob.type,
+      .input_fields = blob.input_fields,
+      .snapshot_id = blob.snapshot_id,
+      .sequence_number = blob.sequence_number,
+      .offset = puffin::PuffinFormat::kMagicLength,
+      .length = static_cast<int64_t>(blob.data.size()),
+      .compression_codec = std::string(puffin::CodecName(codec)),
+      .properties = blob.properties,
+  };
+  const std::string footer =
+      puffin::ToJsonString(puffin::FileMetadata{.blobs = {blob_metadata}});
+
+  std::vector<std::byte> file_data;
+  auto append = [&file_data](const void* data, size_t size) {
+    auto bytes = std::span(reinterpret_cast<const std::byte*>(data), size);
+    file_data.insert(file_data.end(), bytes.begin(), bytes.end());
+  };
+  append(puffin::PuffinFormat::kMagicV1.data(), puffin::PuffinFormat::kMagicV1.size());
+  append(blob.data.data(), blob.data.size());
+  append(puffin::PuffinFormat::kMagicV1.data(), puffin::PuffinFormat::kMagicV1.size());
+  append(footer.data(), footer.size());
+
+  std::array<std::byte, puffin::PuffinFormat::kFooterStructLength> footer_struct{};
+  WriteLittleEndian(static_cast<int32_t>(footer.size()), footer_struct.data());
+  std::memcpy(footer_struct.data() + puffin::PuffinFormat::kFooterStructMagicOffset,
+              puffin::PuffinFormat::kMagicV1.data(),
+              puffin::PuffinFormat::kMagicV1.size());
+  append(footer_struct.data(), footer_struct.size());
+  io->AddFile(path, file_data);
+
+  auto delete_file = std::make_shared<DataFile>(DataFile{
+      .content = DataFile::Content::kPositionDeletes,
+      .file_path = path,
+      .file_format = FileFormatType::kPuffin,
+      .record_count = 3,
+      .file_size_in_bytes = static_cast<int64_t>(file_data.size()),
+      .referenced_data_file = std::string(kReferencedDataFile),
+      .content_offset = blob_metadata.offset,
+      .content_size_in_bytes = blob_metadata.length,
+  });
+  if (test_case.mutate_manifest) {
+    test_case.mutate_manifest(*delete_file);
+  }
+  return delete_file;
+}
+
+}  // namespace
+
+TEST(DVUtilTest, RejectsMalformedPuffinMetadata) {
+  const std::vector<MetadataTestCase> cases = {
+      {
+          .name = "wrong-offset",
+          .mutate_manifest = [](DataFile& file) { ++*file.content_offset; },
+          .expected_error = "No Puffin blob starts",
+      },
+      {
+          .name = "wrong-length",
+          .mutate_manifest = [](DataFile& file) { --*file.content_size_in_bytes; },
+          .expected_error = "manifest content_size_in_bytes",
+      },
+      {
+          .name = "missing-manifest-reference",
+          .mutate_manifest = [](DataFile& file) { file.referenced_data_file.reset(); },
+          .expected_error = "requires referenced_data_file",
+      },
+      {
+          .name = "empty-manifest-reference",
+          .mutate_manifest = [](DataFile& file) { file.referenced_data_file = ""; },
+          .expected_error = "requires referenced_data_file",
+      },
+      {
+          .name = "wrong-type",
+          .mutate_blob = [](puffin::Blob& blob) { blob.type = "test-blob"; },
+          .expected_error = "Invalid deletion vector blob type",
+      },
+      {
+          .name = "wrong-snapshot",
+          .mutate_blob = [](puffin::Blob& blob) { blob.snapshot_id = 0; },
+          .expected_error = "snapshot-id -1",
+      },
+      {
+          .name = "wrong-sequence",
+          .mutate_blob = [](puffin::Blob& blob) { blob.sequence_number = 0; },
+          .expected_error = "sequence-number -1",
+      },
+      {
+          .name = "compressed",
+          .mutate_blob =
+              [](puffin::Blob& blob) {
+                blob.requested_compression = puffin::PuffinCompressionCodec::kZstd;
+              },
+          .expected_error = "must not be compressed",
+      },
+      {
+          .name = "missing-reference",
+          .mutate_blob =
+              [](puffin::Blob& blob) {
+                blob.properties.erase(std::string(kReferencedDataFileProperty));
+              },
+          .expected_error = "requires non-empty 'referenced-data-file'",
+      },
+      {
+          .name = "empty-reference",
+          .mutate_blob =
+              [](puffin::Blob& blob) {
+                blob.properties[std::string(kReferencedDataFileProperty)] = "";
+              },
+          .expected_error = "requires non-empty 'referenced-data-file'",
+      },
+      {
+          .name = "mismatched-reference",
+          .mutate_blob =
+              [](puffin::Blob& blob) {
+                blob.properties[std::string(kReferencedDataFileProperty)] =
+                    "other.parquet";
+              },
+          .expected_error = "does not match Puffin",
+      },
+      {
+          .name = "missing-cardinality",
+          .mutate_blob =
+              [](puffin::Blob& blob) {
+                blob.properties.erase(std::string(kCardinalityProperty));
+              },
+          .expected_error = "requires 'cardinality'",
+      },
+      {
+          .name = "invalid-cardinality",
+          .mutate_blob =
+              [](puffin::Blob& blob) {
+                blob.properties[std::string(kCardinalityProperty)] = "three";
+              },
+          .expected_error = "Failed to parse",
+      },
+      {
+          .name = "negative-cardinality",
+          .mutate_blob =
+              [](puffin::Blob& blob) {
+                blob.properties[std::string(kCardinalityProperty)] = "-1";
+              },
+          .expected_error = "must be non-negative",
+      },
+      {
+          .name = "mismatched-cardinality",
+          .mutate_blob =
+              [](puffin::Blob& blob) {
+                blob.properties[std::string(kCardinalityProperty)] = "4";
+              },
+          .expected_error = "Manifest record_count 3",
+      },
+      {
+          .name = "trailing-bitmap-data",
+          .mutate_data = AddTrailingBitmapData,
+          .expected_error = "Trailing data after bitmaps",
+      },
+  };
+
+  for (const auto& test_case : cases) {
+    SCOPED_TRACE(test_case.name);
+    auto io = std::make_shared<MockFileIO>();
+    ICEBERG_UNWRAP_OR_FAIL(auto delete_file, WriteDVFixture(io, test_case));
+    EXPECT_THAT(DVUtil::ReadDV(delete_file, io),
+                HasErrorMessage(test_case.expected_error));
+  }
+}
+
+TEST(DVUtilTest, ReadsValidPuffinMetadata) {
+  auto io = std::make_shared<MockFileIO>();
+  MetadataTestCase test_case{.name = "valid"};
+  ICEBERG_UNWRAP_OR_FAIL(auto delete_file, WriteDVFixture(io, test_case));
+  ICEBERG_UNWRAP_OR_FAIL(auto positions, DVUtil::ReadDV(delete_file, io));
+
+  EXPECT_EQ(positions.Cardinality(), 3);
+  EXPECT_TRUE(positions.IsDeleted(1));
+  EXPECT_TRUE(positions.IsDeleted(3));
+  EXPECT_TRUE(positions.IsDeleted(5));
+}
+
+TEST(DVUtilTest, ReadsUsingActualFileSize) {
+  auto io = std::make_shared<MockFileIO>();
+  MetadataTestCase test_case{
+      .name = "stale-manifest-file-size",
+      .mutate_manifest = [](DataFile& file) { file.file_size_in_bytes = 0; },
+  };
+  ICEBERG_UNWRAP_OR_FAIL(auto delete_file, WriteDVFixture(io, test_case));
+  ICEBERG_UNWRAP_OR_FAIL(auto positions, DVUtil::ReadDV(delete_file, io));
+
+  EXPECT_EQ(positions.Cardinality(), 3);
+}
+
+}  // namespace iceberg
diff --git a/src/iceberg/test/puffin_dv_interop_test.cc b/src/iceberg/test/puffin_dv_interop_test.cc
index b81dcdd51..b322fe0ae 100644
--- a/src/iceberg/test/puffin_dv_interop_test.cc
+++ b/src/iceberg/test/puffin_dv_interop_test.cc
@@ -17,17 +17,16 @@
  * under the License.
  */
 
-#include <cstddef>
 #include <cstdint>
 #include <filesystem>
 #include <memory>
-#include <span>
 #include <string>
 #include <unordered_set>
 #include <vector>
 
 #include <gtest/gtest.h>
 
+#include "iceberg/deletes/dv_util_internal.h"
 #include "iceberg/deletes/position_delete_index.h"
 #include "iceberg/file_format.h"
 #include "iceberg/manifest/manifest_entry.h"
@@ -100,15 +99,7 @@ void AssertPositions(const ExpectedBlob& expected,
                      const std::shared_ptr<FileIO>& io) {
   ASSERT_TRUE(delete_file->content_offset.has_value());
   ASSERT_TRUE(delete_file->content_size_in_bytes.has_value());
-  ICEBERG_UNWRAP_OR_FAIL(auto input_file, io->NewInputFile(delete_file->file_path));
-  ICEBERG_UNWRAP_OR_FAIL(auto stream, input_file->Open());
-  std::vector<std::byte> data(static_cast<size_t>(*delete_file->content_size_in_bytes));
-  ASSERT_THAT(stream->ReadFully(*delete_file->content_offset, data), IsOk());
-
-  std::span<const uint8_t> blob(reinterpret_cast<const uint8_t*>(data.data()),
-                                data.size());
-  ICEBERG_UNWRAP_OR_FAIL(auto positions,
-                         PositionDeleteIndex::Deserialize(blob, delete_file));
+  ICEBERG_UNWRAP_OR_FAIL(auto positions, DVUtil::ReadDV(delete_file, io));
   EXPECT_EQ(positions.Cardinality(), expected.cardinality);
   int64_t expected_cardinality = static_cast<int64_t>(expected.positions.size());
   for (const auto& range : expected.ranges) {

From 477a2ff8223a77e5152667b8f400b602d8739479 Mon Sep 17 00:00:00 2001
From: xuanyili <xuanyili@uber.com>
Date: Sat, 5 Sep 2026 21:32:08 +0000
Subject: [PATCH 4/4] feat: add table-aware position delete updates

---
 src/iceberg/CMakeLists.txt                    |   1 +
 src/iceberg/data/meson.build                  |   1 +
 src/iceberg/data/position_delete_update.cc    | 324 ++++++++++
 src/iceberg/data/position_delete_update.h     |  60 ++
 src/iceberg/meson.build                       |   1 +
 src/iceberg/test/CMakeLists.txt               |   1 +
 .../test/position_delete_update_test.cc       | 605 ++++++++++++++++++
 7 files changed, 993 insertions(+)
 create mode 100644 src/iceberg/data/position_delete_update.cc
 create mode 100644 src/iceberg/data/position_delete_update.h
 create mode 100644 src/iceberg/test/position_delete_update_test.cc

diff --git a/src/iceberg/CMakeLists.txt b/src/iceberg/CMakeLists.txt
index 8a98274ff..fbb73a22d 100644
--- a/src/iceberg/CMakeLists.txt
+++ b/src/iceberg/CMakeLists.txt
@@ -242,6 +242,7 @@ set(ICEBERG_DATA_SOURCES
     data/delete_loader.cc
     data/equality_delete_writer.cc
     data/file_scan_task_reader.cc
+    data/position_delete_update.cc
     data/position_delete_writer.cc
     data/writer.cc)
 
diff --git a/src/iceberg/data/meson.build b/src/iceberg/data/meson.build
index bbb26db27..6cc94647d 100644
--- a/src/iceberg/data/meson.build
+++ b/src/iceberg/data/meson.build
@@ -22,6 +22,7 @@ install_headers(
         'delete_loader.h',
         'equality_delete_writer.h',
         'file_scan_task_reader.h',
+        'position_delete_update.h',
         'position_delete_writer.h',
         'writer.h',
     ],
diff --git a/src/iceberg/data/position_delete_update.cc b/src/iceberg/data/position_delete_update.cc
new file mode 100644
index 000000000..e206de7bc
--- /dev/null
+++ b/src/iceberg/data/position_delete_update.cc
@@ -0,0 +1,324 @@
+/*
+ * 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.
+ */
+
+#include "iceberg/data/position_delete_update.h"
+
+#include <algorithm>
+#include <format>
+#include <iterator>
+#include <map>
+#include <memory>
+#include <optional>
+#include <set>
+#include <string>
+#include <string_view>
+#include <tuple>
+#include <unordered_map>
+#include <utility>
+#include <vector>
+
+#include "iceberg/data/delete_loader.h"
+#include "iceberg/data/position_delete_writer.h"
+#include "iceberg/deletes/dv_writer.h"
+#include "iceberg/file_format.h"
+#include "iceberg/file_io.h"
+#include "iceberg/location_provider.h"
+#include "iceberg/manifest/manifest_entry.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/schema.h"
+#include "iceberg/table.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/table_properties.h"
+#include "iceberg/table_scan.h"
+#include "iceberg/update/row_delta.h"
+#include "iceberg/util/macros.h"
+#include "iceberg/util/string_util.h"
+#include "iceberg/util/uuid.h"
+
+namespace iceberg {
+
+namespace {
+
+struct TargetFile {
+  std::shared_ptr<DataFile> data_file;
+  std::shared_ptr<PartitionSpec> spec;
+  std::vector<std::shared_ptr<DataFile>> position_delete_files;
+};
+
+struct PreparedDeletes {
+  std::vector<std::shared_ptr<DataFile>> added_files;
+  std::vector<std::shared_ptr<DataFile>> rewritten_files;
+  std::vector<std::string> referenced_data_files;
+};
+
+}  // namespace
+
+class PositionDeleteUpdate::Impl {
+ public:
+  explicit Impl(std::shared_ptr<Table> table) : table_(std::move(table)) {}
+
+  Status Delete(std::string_view data_file_path, int64_t pos) {
+    ICEBERG_PRECHECK(!terminal_, "Position delete update is no longer usable");
+    ICEBERG_PRECHECK(!data_file_path.empty(), "Data file path cannot be empty");
+    ICEBERG_PRECHECK(pos >= 0, "Position delete must be non-negative: {}", pos);
+    deletes_[std::string(data_file_path)].push_back(pos);
+    return {};
+  }
+
+  Status Commit() {
+    ICEBERG_PRECHECK(!terminal_, "Position delete update is no longer usable");
+    ICEBERG_PRECHECK(!deletes_.empty(), "Position delete update is empty");
+    ICEBERG_PRECHECK(table_->metadata()->format_version >= 2,
+                     "Position deletes require table format version 2 or later");
+    ICEBERG_RETURN_UNEXPECTED(CleanupOutput());
+
+    ICEBERG_ASSIGN_OR_RAISE(auto snapshot, table_->current_snapshot());
+    ICEBERG_ASSIGN_OR_RAISE(auto targets, ResolveTargets());
+
+    auto prepared = table_->metadata()->format_version >= 3
+                        ? WriteDeletionVectors(targets)
+                        : WriteParquetDeletes(targets);
+    if (!prepared.has_value()) {
+      return FailAfterCleanup(std::move(prepared.error()));
+    }
+
+    auto row_delta_result = table_->NewRowDelta();
+    if (!row_delta_result.has_value()) {
+      return FailAfterCleanup(std::move(row_delta_result.error()));
+    }
+    auto row_delta = std::move(row_delta_result.value());
+    row_delta->ValidateFromSnapshot(snapshot->snapshot_id)
+        .ValidateDataFilesExist(prepared->referenced_data_files)
+        .ValidateDeletedFiles();
+    for (const auto& file : prepared->added_files) {
+      row_delta->AddDeletes(file);
+    }
+    for (const auto& file : prepared->rewritten_files) {
+      row_delta->RemoveDeletes(file);
+    }
+
+    auto status = row_delta->Commit();
+    if (!status.has_value()) {
+      if (status.error().kind == ErrorKind::kCommitStateUnknown) {
+        terminal_ = true;
+        output_paths_.clear();
+      } else {
+        return FailAfterCleanup(std::move(status.error()));
+      }
+      return status;
+    }
+
+    terminal_ = true;
+    output_paths_.clear();
+    return {};
+  }
+
+ private:
+  Result<std::unordered_map<std::string, TargetFile>> ResolveTargets() const {
+    ICEBERG_ASSIGN_OR_RAISE(auto scan_builder, table_->NewScan());
+    ICEBERG_ASSIGN_OR_RAISE(auto scan, scan_builder->Build());
+    ICEBERG_ASSIGN_OR_RAISE(auto tasks, scan->PlanFiles());
+
+    std::unordered_map<std::string, TargetFile> targets;
+    for (const auto& task : tasks) {
+      const auto& data_file = task->data_file();
+      if (!deletes_.contains(data_file->file_path)) {
+        continue;
+      }
+
+      ICEBERG_PRECHECK(data_file->partition_spec_id.has_value(),
+                       "Data file is missing partition spec ID: {}",
+                       data_file->file_path);
+      ICEBERG_ASSIGN_OR_RAISE(auto spec, table_->metadata()->PartitionSpecById(
+                                             *data_file->partition_spec_id));
+
+      TargetFile target{.data_file = data_file, .spec = std::move(spec)};
+      for (const auto& delete_file : task->delete_files()) {
+        if (delete_file->content == DataFile::Content::kPositionDeletes) {
+          target.position_delete_files.push_back(delete_file);
+        }
+      }
+
+      auto [_, inserted] = targets.emplace(data_file->file_path, std::move(target));
+      ICEBERG_PRECHECK(inserted, "Duplicate live data file path: {}",
+                       data_file->file_path);
+    }
+
+    for (const auto& [path, _] : deletes_) {
+      ICEBERG_PRECHECK(targets.contains(path), "Cannot find live data file: {}", path);
+    }
+    return targets;
+  }
+
+  Result<PreparedDeletes> WriteDeletionVectors(
+      const std::unordered_map<std::string, TargetFile>& targets) {
+    ICEBERG_ASSIGN_OR_RAISE(auto location_provider, table_->location_provider());
+    auto output_path = location_provider->NewDataLocation(
+        std::format("position-deletes-{}.puffin", Uuid::GenerateV7().ToString()));
+    output_paths_.insert(output_path);
+
+    DeleteLoader loader(table_->io());
+    ICEBERG_ASSIGN_OR_RAISE(
+        auto writer,
+        DVWriter::Make(DVWriterOptions{
+            .path = output_path,
+            .io = table_->io(),
+            .load_previous_deletes = [&targets, &loader](std::string_view path)
+                -> Result<std::optional<PositionDeleteIndex>> {
+              const auto target = targets.find(std::string(path));
+              ICEBERG_CHECK(target != targets.end(),
+                            "Missing target data file while loading deletes: {}", path);
+              if (target->second.position_delete_files.empty()) {
+                return std::nullopt;
+              }
+              ICEBERG_ASSIGN_OR_RAISE(
+                  auto index,
+                  loader.LoadPositionDeletes(target->second.position_delete_files, path));
+              return std::optional<PositionDeleteIndex>(std::move(index));
+            },
+        }));
+
+    for (const auto& [path, positions] : deletes_) {
+      const auto& target = targets.at(path);
+      for (int64_t pos : positions) {
+        ICEBERG_RETURN_UNEXPECTED(
+            writer->Delete(path, pos, target.spec, target.data_file->partition));
+      }
+    }
+    ICEBERG_RETURN_UNEXPECTED(writer->Close());
+    ICEBERG_ASSIGN_OR_RAISE(auto result, writer->Metadata());
+    return PreparedDeletes{
+        .added_files = std::move(result.data_files),
+        .rewritten_files = std::move(result.rewritten_delete_files),
+        .referenced_data_files = std::move(result.referenced_data_files),
+    };
+  }
+
+  Result<PreparedDeletes> WriteParquetDeletes(
+      const std::unordered_map<std::string, TargetFile>& targets) {
+    ICEBERG_ASSIGN_OR_RAISE(auto location_provider, table_->location_provider());
+    ICEBERG_ASSIGN_OR_RAISE(auto schema, table_->schema());
+
+    PreparedDeletes result;
+    result.added_files.reserve(deletes_.size());
+    result.referenced_data_files.reserve(deletes_.size());
+    auto properties = table_->properties().configs();
+    properties[TableProperties::kParquetCompression.key()] =
+        table_->properties().Get(TableProperties::kDeleteParquetCompression);
+    properties[TableProperties::kParquetCompressionLevel.key()] =
+        table_->properties().Get(TableProperties::kDeleteParquetCompressionLevel);
+    const auto write_uuid = Uuid::GenerateV7().ToString();
+    size_t file_number = 0;
+
+    for (const auto& [path, positions] : deletes_) {
+      auto sorted_positions = positions;
+      std::ranges::sort(sorted_positions);
+
+      const auto& target = targets.at(path);
+      const auto filename =
+          std::format("position-deletes-{}-{}.parquet", write_uuid, file_number++);
+      auto output_path = location_provider->NewDataLocation(filename);
+      output_paths_.insert(output_path);
+
+      ICEBERG_ASSIGN_OR_RAISE(auto writer,
+                              PositionDeleteWriter::Make(PositionDeleteWriterOptions{
+                                  .path = output_path,
+                                  .schema = schema,
+                                  .spec = target.spec,
+                                  .partition = target.data_file->partition,
+                                  .format = FileFormatType::kParquet,
+                                  .io = table_->io(),
+                                  .properties = properties,
+                              }));
+      for (int64_t pos : sorted_positions) {
+        ICEBERG_RETURN_UNEXPECTED(writer->WriteDelete(path, pos));
+      }
+      ICEBERG_RETURN_UNEXPECTED(writer->Close());
+      ICEBERG_ASSIGN_OR_RAISE(auto metadata, writer->Metadata());
+      result.added_files.insert(result.added_files.end(),
+                                std::make_move_iterator(metadata.data_files.begin()),
+                                std::make_move_iterator(metadata.data_files.end()));
+      result.referenced_data_files.push_back(path);
+    }
+    return result;
+  }
+
+  Status CleanupOutput() {
+    std::optional<Error> first_error;
+    for (auto it = output_paths_.begin(); it != output_paths_.end();) {
+      auto status = table_->io()->DeleteFile(*it);
+      if (status.has_value()) {
+        it = output_paths_.erase(it);
+        continue;
+      }
+
+      if (!first_error.has_value()) {
+        first_error = std::move(status.error());
+      } else {
+        first_error->message += "; additionally failed to delete output file: ";
+        first_error->message += status.error().message;
+      }
+      ++it;
+    }
+    if (first_error.has_value()) {
+      return std::unexpected(std::move(*first_error));
+    }
+    return {};
+  }
+
+  Status FailAfterCleanup(Error error) {
+    auto cleanup_status = CleanupOutput();
+    if (!cleanup_status.has_value()) {
+      error.message += "; additionally failed to clean output files: ";
+      error.message += cleanup_status.error().message;
+    }
+    return std::unexpected(std::move(error));
+  }
+
+  std::shared_ptr<Table> table_;
+  std::map<std::string, std::vector<int64_t>, StringLess> deletes_;
+  std::set<std::string, StringLess> output_paths_;
+  bool terminal_ = false;
+};
+
+PositionDeleteUpdate::PositionDeleteUpdate(std::unique_ptr<Impl> impl)
+    : impl_(std::move(impl)) {}
+
+PositionDeleteUpdate::~PositionDeleteUpdate() = default;
+
+Result<std::unique_ptr<PositionDeleteUpdate>> PositionDeleteUpdate::Make(
+    std::shared_ptr<Table> table) {
+  ICEBERG_PRECHECK(table != nullptr,
+                   "Cannot create position delete update without table");
+  return std::unique_ptr<PositionDeleteUpdate>(
+      new PositionDeleteUpdate(std::make_unique<Impl>(std::move(table))));
+}
+
+PositionDeleteUpdate& PositionDeleteUpdate::Delete(std::string_view data_file_path,
+                                                   int64_t pos) {
+  ICEBERG_BUILDER_RETURN_IF_ERROR(impl_->Delete(data_file_path, pos));
+  return *this;
+}
+
+Status PositionDeleteUpdate::Commit() {
+  ICEBERG_RETURN_UNEXPECTED(CheckErrors());
+  return impl_->Commit();
+}
+
+}  // namespace iceberg
diff --git a/src/iceberg/data/position_delete_update.h b/src/iceberg/data/position_delete_update.h
new file mode 100644
index 000000000..3b531d8fa
--- /dev/null
+++ b/src/iceberg/data/position_delete_update.h
@@ -0,0 +1,60 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+/// \file iceberg/data/position_delete_update.h
+/// Table-aware position delete writing and commit.
+
+#include <cstdint>
+#include <memory>
+#include <string_view>
+
+#include "iceberg/iceberg_data_export.h"
+#include "iceberg/result.h"
+#include "iceberg/type_fwd.h"
+#include "iceberg/util/error_collector.h"
+
+namespace iceberg {
+
+/// \brief Writes and commits position deletes using the table format.
+///
+/// Format v2 tables produce Parquet position delete files. Format v3 tables
+/// produce deletion vectors and merge previous file-scoped position deletes.
+class ICEBERG_DATA_EXPORT PositionDeleteUpdate : public ErrorCollector {
+ public:
+  ~PositionDeleteUpdate() override;
+
+  /// \brief Create a position delete update for a table.
+  static Result<std::unique_ptr<PositionDeleteUpdate>> Make(std::shared_ptr<Table> table);
+
+  /// \brief Add a deleted row position for a live data file.
+  PositionDeleteUpdate& Delete(std::string_view data_file_path, int64_t pos);
+
+  /// \brief Write and commit all added position deletes.
+  Status Commit();
+
+ private:
+  class Impl;
+  std::unique_ptr<Impl> impl_;
+
+  explicit PositionDeleteUpdate(std::unique_ptr<Impl> impl);
+};
+
+}  // namespace iceberg
diff --git a/src/iceberg/meson.build b/src/iceberg/meson.build
index 989f4ae03..7ecb5e640 100644
--- a/src/iceberg/meson.build
+++ b/src/iceberg/meson.build
@@ -223,6 +223,7 @@ iceberg_data_sources = files(
     'data/delete_loader.cc',
     'data/equality_delete_writer.cc',
     'data/file_scan_task_reader.cc',
+    'data/position_delete_update.cc',
     'data/position_delete_writer.cc',
     'data/writer.cc',
 )
diff --git a/src/iceberg/test/CMakeLists.txt b/src/iceberg/test/CMakeLists.txt
index 3238e9c2d..960534c7e 100644
--- a/src/iceberg/test/CMakeLists.txt
+++ b/src/iceberg/test/CMakeLists.txt
@@ -261,6 +261,7 @@ if(ICEBERG_BUILD_BUNDLE)
                    delete_loader_test.cc
                    dv_writer_test.cc
                    file_scan_task_reader_test.cc
+                   position_delete_update_test.cc
                    literal_util_test.cc)
 
 endif()
diff --git a/src/iceberg/test/position_delete_update_test.cc b/src/iceberg/test/position_delete_update_test.cc
new file mode 100644
index 000000000..8bf2e3d4f
--- /dev/null
+++ b/src/iceberg/test/position_delete_update_test.cc
@@ -0,0 +1,605 @@
+/*
+ * 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.
+ */
+
+#include "iceberg/data/position_delete_update.h"
+
+#include <algorithm>
+#include <cstdint>
+#include <format>
+#include <iterator>
+#include <memory>
+#include <span>
+#include <string>
+#include <vector>
+
+#include <arrow/array.h>
+#include <arrow/c/bridge.h>
+#include <arrow/filesystem/filesystem.h>
+#include <gmock/gmock.h>
+#include <gtest/gtest.h>
+
+#include "iceberg/arrow/arrow_io_internal.h"
+#include "iceberg/avro/avro_register.h"
+#include "iceberg/data/delete_loader.h"
+#include "iceberg/data/position_delete_writer.h"
+#include "iceberg/deletes/position_delete_index.h"
+#include "iceberg/file_reader.h"
+#include "iceberg/manifest/manifest_reader.h"
+#include "iceberg/metadata_columns.h"
+#include "iceberg/parquet/parquet_register.h"
+#include "iceberg/partition_spec.h"
+#include "iceberg/schema.h"
+#include "iceberg/schema_field.h"
+#include "iceberg/schema_internal.h"
+#include "iceberg/snapshot.h"
+#include "iceberg/table.h"
+#include "iceberg/table_metadata.h"
+#include "iceberg/table_properties.h"
+#include "iceberg/table_scan.h"
+#include "iceberg/test/matchers.h"
+#include "iceberg/test/mock_catalog.h"
+#include "iceberg/test/update_test_base.h"
+#include "iceberg/update/fast_append.h"
+#include "iceberg/update/row_delta.h"
+#include "iceberg/update/update_partition_spec.h"
+#include "iceberg/update/update_properties.h"
+#include "iceberg/util/uuid.h"
+
+namespace iceberg {
+
+namespace {
+
+struct RoutingCase {
+  int8_t format_version;
+  bool unpartitioned;
+  FileFormatType expected_format;
+};
+
+class FailOncePuffinDeleteFileIO : public arrow::ArrowFileSystemFileIO {
+ public:
+  explicit FailOncePuffinDeleteFileIO(
+      std::shared_ptr<::arrow::fs::FileSystem> file_system)
+      : ArrowFileSystemFileIO(std::move(file_system)) {}
+
+  Status DeleteFile(const std::string& file_location) override {
+    if (!file_location.ends_with(".puffin")) {
+      return ArrowFileSystemFileIO::DeleteFile(file_location);
+    }
+
+    puffin_delete_attempts.push_back(file_location);
+    if (fail_next_puffin_delete_) {
+      fail_next_puffin_delete_ = false;
+      return IOError("injected cleanup failure for {}", file_location);
+    }
+    return ArrowFileSystemFileIO::DeleteFile(file_location);
+  }
+
+  std::vector<std::string> puffin_delete_attempts;
+
+ private:
+  bool fail_next_puffin_delete_ = true;
+};
+
+class PositionDeleteUpdateTest : public MinimalUpdateTestBase,
+                                 public ::testing::WithParamInterface<RoutingCase> {
+ protected:
+  static void SetUpTestSuite() {
+    avro::RegisterAll();
+    parquet::RegisterAll();
+  }
+
+  int8_t format_version() const override { return GetParam().format_version; }
+
+  void SetUp() override {
+    MinimalUpdateTestBase::SetUp();
+    if (GetParam().unpartitioned) {
+      RegisterUnpartitionedTable();
+    }
+    if (GetParam().format_version == 2) {
+      ICEBERG_UNWRAP_OR_FAIL(auto properties, table_->NewUpdateProperties());
+      properties->Set(TableProperties::kDeleteParquetCompression.key(), "uncompressed");
+      ASSERT_THAT(properties->Commit(), IsOk());
+      ASSERT_THAT(table_->Refresh(), IsOk());
+    }
+    ICEBERG_UNWRAP_OR_FAIL(spec_, table_->spec());
+    data_file_ = MakeDataFile();
+    AppendDataFile();
+  }
+
+  void RegisterUnpartitionedTable() {
+    ICEBERG_UNWRAP_OR_FAIL(
+        auto metadata, ReadTableMetadataFromResource("TableMetadataV3ValidMinimal.json"));
+    metadata->location = table_location_;
+    metadata->partition_specs = {PartitionSpec::Unpartitioned()};
+    metadata->default_spec_id = PartitionSpec::kInitialSpecId;
+
+    const auto metadata_location =
+        std::format("{}/metadata/00001-{}.metadata.json", table_location_,
+                    Uuid::GenerateV7().ToString());
+    ASSERT_THAT(TableMetadataUtil::Write(*file_io_, metadata_location, *metadata),
+                IsOk());
+    ASSERT_THAT(catalog_->DropTable(table_ident_, /*purge=*/false), IsOk());
+    ICEBERG_UNWRAP_OR_FAIL(table_,
+                           catalog_->RegisterTable(table_ident_, metadata_location));
+  }
+
+  std::shared_ptr<DataFile> MakeDataFile() const {
+    auto file = std::make_shared<DataFile>();
+    file->content = DataFile::Content::kData;
+    file->file_path = table_location_ + "/data/file.parquet";
+    file->file_format = FileFormatType::kParquet;
+    file->partition = GetParam().unpartitioned ? PartitionValues{}
+                                               : PartitionValues({Literal::Long(10)});
+    file->file_size_in_bytes = 1024;
+    file->record_count = 10;
+    file->partition_spec_id = spec_->spec_id();
+    return file;
+  }
+
+  void AppendDataFile() {
+    ICEBERG_UNWRAP_OR_FAIL(auto append, table_->NewFastAppend());
+    append->AppendFile(data_file_);
+    ASSERT_THAT(append->Commit(), IsOk());
+    ASSERT_THAT(table_->Refresh(), IsOk());
+  }
+
+  Result<std::shared_ptr<FileScanTask>> CurrentTask() {
+    ICEBERG_ASSIGN_OR_RAISE(auto builder, table_->NewScan());
+    ICEBERG_ASSIGN_OR_RAISE(auto scan, builder->Build());
+    ICEBERG_ASSIGN_OR_RAISE(auto tasks, scan->PlanFiles());
+    ICEBERG_CHECK(tasks.size() == 1, "Expected one file scan task, found {}",
+                  tasks.size());
+    return tasks.front();
+  }
+
+  Result<PositionDeleteIndex> LoadPositions(const FileScanTask& task) {
+    std::vector<std::shared_ptr<DataFile>> deletes;
+    std::ranges::copy_if(task.delete_files(), std::back_inserter(deletes),
+                         [](const auto& file) {
+                           return file->content == DataFile::Content::kPositionDeletes;
+                         });
+    DeleteLoader loader(file_io_);
+    return loader.LoadPositionDeletes(deletes, task.data_file()->file_path);
+  }
+
+  std::shared_ptr<PartitionSpec> spec_;
+  std::shared_ptr<DataFile> data_file_;
+};
+
+TEST_P(PositionDeleteUpdateTest, HandlesDescendingInputAndSortsV2Positions) {
+  ICEBERG_UNWRAP_OR_FAIL(auto update, PositionDeleteUpdate::Make(table_));
+  update->Delete(data_file_->file_path, 7).Delete(data_file_->file_path, 2);
+  ASSERT_THAT(update->Commit(), IsOk());
+  ASSERT_THAT(table_->Refresh(), IsOk());
+
+  ICEBERG_UNWRAP_OR_FAIL(auto task, CurrentTask());
+  ASSERT_EQ(task->delete_files().size(), 1);
+  const auto& delete_file = task->delete_files().front();
+  ASSERT_EQ(delete_file->file_format, GetParam().expected_format);
+  if (GetParam().format_version != 2) {
+    return;
+  }
+
+  auto delete_schema = std::make_shared<Schema>(std::vector<SchemaField>{
+      MetadataColumns::kDeleteFilePath, MetadataColumns::kDeleteFilePos});
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto reader,
+      ReaderFactoryRegistry::Open(
+          FileFormatType::kParquet,
+          {.path = delete_file->file_path, .io = file_io_, .projection = delete_schema}));
+  ICEBERG_UNWRAP_OR_FAIL(auto batch, reader->Next());
+  ASSERT_TRUE(batch.has_value());
+
+  ArrowSchema arrow_schema;
+  ASSERT_THAT(ToArrowSchema(*delete_schema, &arrow_schema), IsOk());
+  auto arrow_type = ::arrow::ImportType(&arrow_schema).ValueOrDie();
+  auto rows = ::arrow::ImportArray(&batch.value(), arrow_type).ValueOrDie();
+  auto struct_rows = std::static_pointer_cast<::arrow::StructArray>(rows);
+  auto positions = std::static_pointer_cast<::arrow::Int64Array>(struct_rows->field(1));
+  ASSERT_EQ(positions->length(), 2);
+  EXPECT_EQ(positions->Value(0), 2);
+  EXPECT_EQ(positions->Value(1), 7);
+}
+
+TEST_P(PositionDeleteUpdateTest, RoutesAndCommitsPositionDeletes) {
+  ICEBERG_UNWRAP_OR_FAIL(auto update, PositionDeleteUpdate::Make(table_));
+  update->Delete(data_file_->file_path, 2).Delete(data_file_->file_path, 7);
+  ASSERT_THAT(update->Commit(), IsOk());
+  ASSERT_THAT(table_->Refresh(), IsOk());
+
+  ICEBERG_UNWRAP_OR_FAIL(auto task, CurrentTask());
+  ASSERT_EQ(task->delete_files().size(), 1);
+  const auto& delete_file = task->delete_files().front();
+  EXPECT_EQ(delete_file->file_format, GetParam().expected_format);
+  EXPECT_EQ(delete_file->partition_spec_id, data_file_->partition_spec_id);
+  EXPECT_EQ(delete_file->partition, data_file_->partition);
+
+  ICEBERG_UNWRAP_OR_FAIL(auto positions, LoadPositions(*task));
+  EXPECT_EQ(positions.Cardinality(), 2);
+  EXPECT_TRUE(positions.IsDeleted(2));
+  EXPECT_TRUE(positions.IsDeleted(7));
+}
+
+INSTANTIATE_TEST_SUITE_P(
+    FormatAndPartitioning, PositionDeleteUpdateTest,
+    ::testing::Values(RoutingCase{.format_version = 2,
+                                  .unpartitioned = false,
+                                  .expected_format = FileFormatType::kParquet},
+                      RoutingCase{.format_version = 3,
+                                  .unpartitioned = false,
+                                  .expected_format = FileFormatType::kPuffin},
+                      RoutingCase{.format_version = 3,
+                                  .unpartitioned = true,
+                                  .expected_format = FileFormatType::kPuffin}));
+
+class PositionDeleteV3Test : public MinimalUpdateTestBase {
+ protected:
+  static void SetUpTestSuite() {
+    avro::RegisterAll();
+    parquet::RegisterAll();
+  }
+
+  int8_t format_version() const override { return 3; }
+
+  void SetUp() override {
+    MinimalUpdateTestBase::SetUp();
+    ICEBERG_UNWRAP_OR_FAIL(spec_, table_->spec());
+    ICEBERG_UNWRAP_OR_FAIL(schema_, table_->schema());
+    data_file_ = MakeDataFile();
+    AppendDataFile();
+  }
+
+  std::shared_ptr<DataFile> MakeDataFile() const {
+    auto file = std::make_shared<DataFile>();
+    file->content = DataFile::Content::kData;
+    file->file_path = table_location_ + "/data/file.parquet";
+    file->file_format = FileFormatType::kParquet;
+    file->partition = PartitionValues({Literal::Long(10)});
+    file->file_size_in_bytes = 1024;
+    file->record_count = 10;
+    file->partition_spec_id = spec_->spec_id();
+    return file;
+  }
+
+  void AppendDataFile() {
+    ICEBERG_UNWRAP_OR_FAIL(auto append, table_->NewFastAppend());
+    append->AppendFile(data_file_);
+    ASSERT_THAT(append->Commit(), IsOk());
+    ASSERT_THAT(table_->Refresh(), IsOk());
+  }
+
+  Result<std::shared_ptr<FileScanTask>> CurrentTask(const std::shared_ptr<Table>& table) {
+    ICEBERG_ASSIGN_OR_RAISE(auto builder, table->NewScan());
+    ICEBERG_ASSIGN_OR_RAISE(auto scan, builder->Build());
+    ICEBERG_ASSIGN_OR_RAISE(auto tasks, scan->PlanFiles());
+    ICEBERG_CHECK(tasks.size() == 1, "Expected one file scan task, found {}",
+                  tasks.size());
+    return tasks.front();
+  }
+
+  Result<PositionDeleteIndex> LoadPositions(const FileScanTask& task) {
+    DeleteLoader loader(file_io_);
+    return loader.LoadPositionDeletes(task.delete_files(), task.data_file()->file_path);
+  }
+
+  Result<std::shared_ptr<DataFile>> WritePositionDeletes(
+      std::span<const int64_t> positions) {
+    const auto path = table_location_ + "/data/existing-position-deletes.parquet";
+    ICEBERG_ASSIGN_OR_RAISE(
+        auto writer,
+        PositionDeleteWriter::Make(PositionDeleteWriterOptions{
+            .path = path,
+            .schema = schema_,
+            .spec = spec_,
+            .partition = data_file_->partition,
+            .format = FileFormatType::kParquet,
+            .io = file_io_,
+            .properties = {{"write.parquet.compression-codec", "uncompressed"}},
+        }));
+    for (int64_t pos : positions) {
+      ICEBERG_RETURN_UNEXPECTED(writer->WriteDelete(data_file_->file_path, pos));
+    }
+    ICEBERG_RETURN_UNEXPECTED(writer->Close());
+    ICEBERG_ASSIGN_OR_RAISE(auto result, writer->Metadata());
+    ICEBERG_CHECK(result.data_files.size() == 1,
+                  "Expected one position delete file, found {}",
+                  result.data_files.size());
+    return result.data_files.front();
+  }
+
+  Result<std::vector<ManifestEntry>> CurrentDeleteEntries() {
+    ICEBERG_ASSIGN_OR_RAISE(auto snapshot, table_->current_snapshot());
+    SnapshotCache cache(snapshot.get());
+    ICEBERG_ASSIGN_OR_RAISE(auto manifests, cache.DeleteManifests(file_io_));
+    std::vector<ManifestEntry> entries;
+    for (const auto& manifest : manifests) {
+      ICEBERG_ASSIGN_OR_RAISE(
+          auto spec, table_->metadata()->PartitionSpecById(manifest.partition_spec_id));
+      ICEBERG_ASSIGN_OR_RAISE(
+          auto reader,
+          ManifestReader::Make(manifest, file_io_, schema_, std::move(spec)));
+      ICEBERG_ASSIGN_OR_RAISE(auto manifest_entries, reader->Entries());
+      entries.insert(entries.end(), std::make_move_iterator(manifest_entries.begin()),
+                     std::make_move_iterator(manifest_entries.end()));
+    }
+    return entries;
+  }
+
+  void ConfigureRetries(int32_t retries) {
+    ICEBERG_UNWRAP_OR_FAIL(auto properties, table_->NewUpdateProperties());
+    properties->Set(TableProperties::kCommitNumRetries.key(), std::to_string(retries))
+        .Set(TableProperties::kCommitMinRetryWaitMs.key(), "1")
+        .Set(TableProperties::kCommitMaxRetryWaitMs.key(), "1")
+        .Set(TableProperties::kCommitTotalRetryTimeMs.key(), "1000");
+    ASSERT_THAT(properties->Commit(), IsOk());
+    ASSERT_THAT(table_->Refresh(), IsOk());
+  }
+
+  std::vector<std::string> PuffinFiles() {
+    auto arrow_io = std::dynamic_pointer_cast<arrow::ArrowFileSystemFileIO>(file_io_);
+    EXPECT_NE(arrow_io, nullptr);
+    ::arrow::fs::FileSelector selector;
+    selector.base_dir = table_location_ + "/data";
+    selector.recursive = true;
+    auto infos = arrow_io->fs()->GetFileInfo(selector);
+    EXPECT_TRUE(infos.ok()) << infos.status().ToString();
+    std::vector<std::string> paths;
+    if (!infos.ok()) {
+      return paths;
+    }
+    for (const auto& info : *infos) {
+      if (info.path().ends_with(".puffin")) {
+        paths.push_back(info.path());
+      }
+    }
+    return paths;
+  }
+
+  std::shared_ptr<PartitionSpec> spec_;
+  std::shared_ptr<Schema> schema_;
+  std::shared_ptr<DataFile> data_file_;
+};
+
+TEST_F(PositionDeleteV3Test, SecondDeleteMergesAndSupersedesPreviousDV) {
+  ICEBERG_UNWRAP_OR_FAIL(auto first, PositionDeleteUpdate::Make(table_));
+  first->Delete(data_file_->file_path, 1);
+  ASSERT_THAT(first->Commit(), IsOk());
+  ASSERT_THAT(table_->Refresh(), IsOk());
+  ICEBERG_UNWRAP_OR_FAIL(auto old_task, CurrentTask(table_));
+  const auto old_dv = old_task->delete_files().front();
+
+  ICEBERG_UNWRAP_OR_FAIL(auto second, PositionDeleteUpdate::Make(table_));
+  second->Delete(data_file_->file_path, 3);
+  ASSERT_THAT(second->Commit(), IsOk());
+  ASSERT_THAT(table_->Refresh(), IsOk());
+
+  ICEBERG_UNWRAP_OR_FAIL(auto task, CurrentTask(table_));
+  ASSERT_EQ(task->delete_files().size(), 1);
+  EXPECT_NE(task->delete_files().front()->file_path, old_dv->file_path);
+  ICEBERG_UNWRAP_OR_FAIL(auto positions, LoadPositions(*task));
+  EXPECT_EQ(positions.Cardinality(), 2);
+  EXPECT_TRUE(positions.IsDeleted(1));
+  EXPECT_TRUE(positions.IsDeleted(3));
+
+  ICEBERG_UNWRAP_OR_FAIL(auto entries, CurrentDeleteEntries());
+  EXPECT_TRUE(std::ranges::any_of(entries, [&old_dv](const ManifestEntry& entry) {
+    return entry.status == ManifestStatus::kDeleted && entry.data_file != nullptr &&
+           entry.data_file->file_path == old_dv->file_path &&
+           entry.data_file->content_offset == old_dv->content_offset;
+  }));
+}
+
+TEST_F(PositionDeleteV3Test, MergesAndSupersedesFileScopedParquetDelete) {
+  RegisterTableFromResource("TableMetadataV2ValidMinimal.json");
+  ICEBERG_UNWRAP_OR_FAIL(spec_, table_->spec());
+  ICEBERG_UNWRAP_OR_FAIL(schema_, table_->schema());
+  data_file_ = MakeDataFile();
+  AppendDataFile();
+
+  const std::vector<int64_t> existing_positions{1, 3};
+  ICEBERG_UNWRAP_OR_FAIL(auto old_delete, WritePositionDeletes(existing_positions));
+  ASSERT_EQ(old_delete->referenced_data_file, data_file_->file_path);
+  ICEBERG_UNWRAP_OR_FAIL(auto row_delta, table_->NewRowDelta());
+  row_delta->AddDeletes(old_delete);
+  ASSERT_THAT(row_delta->Commit(), IsOk());
+  ASSERT_THAT(table_->Refresh(), IsOk());
+
+  ICEBERG_UNWRAP_OR_FAIL(auto properties, table_->NewUpdateProperties());
+  properties->Set(TableProperties::kFormatVersion.key(), "3");
+  ASSERT_THAT(properties->Commit(), IsOk());
+  ASSERT_THAT(table_->Refresh(), IsOk());
+
+  ICEBERG_UNWRAP_OR_FAIL(auto update, PositionDeleteUpdate::Make(table_));
+  update->Delete(data_file_->file_path, 5);
+  ASSERT_THAT(update->Commit(), IsOk());
+  ASSERT_THAT(table_->Refresh(), IsOk());
+
+  ICEBERG_UNWRAP_OR_FAIL(auto task, CurrentTask(table_));
+  ASSERT_EQ(task->delete_files().size(), 1);
+  EXPECT_EQ(task->delete_files().front()->file_format, FileFormatType::kPuffin);
+  ICEBERG_UNWRAP_OR_FAIL(auto positions, LoadPositions(*task));
+  EXPECT_EQ(positions.Cardinality(), 3);
+  EXPECT_TRUE(positions.IsDeleted(1));
+  EXPECT_TRUE(positions.IsDeleted(3));
+  EXPECT_TRUE(positions.IsDeleted(5));
+
+  ICEBERG_UNWRAP_OR_FAIL(auto entries, CurrentDeleteEntries());
+  EXPECT_TRUE(std::ranges::any_of(entries, [&old_delete](const ManifestEntry& entry) {
+    return entry.status == ManifestStatus::kDeleted && entry.data_file != nullptr &&
+           entry.data_file->file_path == old_delete->file_path;
+  }));
+}
+
+TEST_F(PositionDeleteV3Test, UsesTargetDataFileSpecAfterPartitionEvolution) {
+  ICEBERG_UNWRAP_OR_FAIL(auto spec_update, table_->NewUpdatePartitionSpec());
+  spec_update->RemoveField("x");
+  ASSERT_THAT(spec_update->Commit(), IsOk());
+  ASSERT_THAT(table_->Refresh(), IsOk());
+  ASSERT_NE(table_->metadata()->default_spec_id, data_file_->partition_spec_id);
+
+  ICEBERG_UNWRAP_OR_FAIL(auto update, PositionDeleteUpdate::Make(table_));
+  update->Delete(data_file_->file_path, 4);
+  ASSERT_THAT(update->Commit(), IsOk());
+  ASSERT_THAT(table_->Refresh(), IsOk());
+
+  ICEBERG_UNWRAP_OR_FAIL(auto task, CurrentTask(table_));
+  const auto& dv = task->delete_files().front();
+  EXPECT_EQ(dv->partition_spec_id, data_file_->partition_spec_id);
+  EXPECT_EQ(dv->partition, data_file_->partition);
+}
+
+TEST_F(PositionDeleteV3Test, CommitRetryReusesWrittenDV) {
+  ConfigureRetries(1);
+  auto mock_catalog = std::make_shared<::testing::NiceMock<MockCatalog>>();
+  std::weak_ptr<Catalog> weak_catalog = mock_catalog;
+  int update_calls = 0;
+  ON_CALL(*mock_catalog, LoadTable(::testing::_))
+      .WillByDefault(
+          [this, weak_catalog](const TableIdentifier&) -> Result<std::shared_ptr<Table>> {
+            auto retry_catalog = weak_catalog.lock();
+            ICEBERG_CHECK(retry_catalog != nullptr, "Mock catalog expired");
+            ICEBERG_ASSIGN_OR_RAISE(auto loaded, catalog_->LoadTable(table_ident_));
+            return Table::Make(loaded->name(), loaded->metadata(),
+                               std::string(loaded->metadata_file_location()),
+                               loaded->io(), std::move(retry_catalog));
+          });
+  ON_CALL(*mock_catalog, UpdateTable(::testing::_, ::testing::_, ::testing::_))
+      .WillByDefault(
+          [this, &update_calls, weak_catalog](
+              const TableIdentifier& identifier,
+              const std::vector<std::unique_ptr<TableRequirement>>& requirements,
+              const std::vector<std::unique_ptr<TableUpdate>>& updates)
+              -> Result<std::shared_ptr<Table>> {
+            if (++update_calls == 1) {
+              return CommitFailed("injected conflict");
+            }
+            ICEBERG_ASSIGN_OR_RAISE(
+                auto committed, catalog_->UpdateTable(identifier, requirements, updates));
+            auto retry_catalog = weak_catalog.lock();
+            ICEBERG_CHECK(retry_catalog != nullptr, "Mock catalog expired");
+            return Table::Make(committed->name(), committed->metadata(),
+                               std::string(committed->metadata_file_location()),
+                               committed->io(), std::move(retry_catalog));
+          });
+  ICEBERG_UNWRAP_OR_FAIL(auto mock_table,
+                         Table::Make(table_->name(), table_->metadata(),
+                                     std::string(table_->metadata_file_location()),
+                                     table_->io(), mock_catalog));
+
+  ICEBERG_UNWRAP_OR_FAIL(auto update, PositionDeleteUpdate::Make(mock_table));
+  update->Delete(data_file_->file_path, 5);
+  ASSERT_THAT(update->Commit(), IsOk());
+
+  EXPECT_EQ(update_calls, 2);
+  EXPECT_EQ(PuffinFiles().size(), 1);
+}
+
+TEST_F(PositionDeleteV3Test, FailedCommitCleansWrittenDV) {
+  ConfigureRetries(0);
+  auto mock_catalog = std::make_shared<::testing::NiceMock<MockCatalog>>();
+  ON_CALL(*mock_catalog, UpdateTable(::testing::_, ::testing::_, ::testing::_))
+      .WillByDefault([](const TableIdentifier&,
+                        const std::vector<std::unique_ptr<TableRequirement>>&,
+                        const std::vector<std::unique_ptr<TableUpdate>>&)
+                         -> Result<std::shared_ptr<Table>> {
+        return CommitFailed("injected failure");
+      });
+  ICEBERG_UNWRAP_OR_FAIL(auto mock_table,
+                         Table::Make(table_->name(), table_->metadata(),
+                                     std::string(table_->metadata_file_location()),
+                                     table_->io(), mock_catalog));
+
+  ICEBERG_UNWRAP_OR_FAIL(auto update, PositionDeleteUpdate::Make(mock_table));
+  update->Delete(data_file_->file_path, 6);
+  EXPECT_THAT(update->Commit(), IsError(ErrorKind::kCommitFailed));
+  EXPECT_TRUE(PuffinFiles().empty());
+}
+
+TEST_F(PositionDeleteV3Test, CleanupFailureIsReportedAndRetried) {
+  ConfigureRetries(0);
+  auto mock_catalog = std::make_shared<::testing::NiceMock<MockCatalog>>();
+  int update_calls = 0;
+  ON_CALL(*mock_catalog, UpdateTable(::testing::_, ::testing::_, ::testing::_))
+      .WillByDefault(
+          [this, &update_calls](
+              const TableIdentifier& identifier,
+              const std::vector<std::unique_ptr<TableRequirement>>& requirements,
+              const std::vector<std::unique_ptr<TableUpdate>>& updates)
+              -> Result<std::shared_ptr<Table>> {
+            if (++update_calls == 1) {
+              return CommitFailed("injected commit failure");
+            }
+            return catalog_->UpdateTable(identifier, requirements, updates);
+          });
+  auto arrow_io = std::dynamic_pointer_cast<arrow::ArrowFileSystemFileIO>(file_io_);
+  ASSERT_NE(arrow_io, nullptr);
+  auto failing_io = std::make_shared<FailOncePuffinDeleteFileIO>(arrow_io->fs());
+  ICEBERG_UNWRAP_OR_FAIL(auto mock_table,
+                         Table::Make(table_->name(), table_->metadata(),
+                                     std::string(table_->metadata_file_location()),
+                                     failing_io, mock_catalog));
+
+  ICEBERG_UNWRAP_OR_FAIL(auto update, PositionDeleteUpdate::Make(mock_table));
+  update->Delete(data_file_->file_path, 6);
+  auto first_status = update->Commit();
+  EXPECT_THAT(first_status, IsError(ErrorKind::kCommitFailed));
+  EXPECT_THAT(first_status, HasErrorMessage("injected commit failure"));
+  EXPECT_THAT(first_status, HasErrorMessage("injected cleanup failure"));
+  ASSERT_EQ(PuffinFiles().size(), 1);
+  ASSERT_EQ(failing_io->puffin_delete_attempts.size(), 1);
+  const auto retained_path = failing_io->puffin_delete_attempts.front();
+
+  EXPECT_THAT(update->Commit(), IsOk());
+  EXPECT_EQ(update_calls, 2);
+  EXPECT_EQ(PuffinFiles().size(), 1);
+  ASSERT_EQ(failing_io->puffin_delete_attempts.size(), 2);
+  EXPECT_EQ(std::ranges::count(failing_io->puffin_delete_attempts, retained_path), 2);
+}
+
+TEST_F(PositionDeleteV3Test, CommitStateUnknownRelinquishesOutputOwnership) {
+  ConfigureRetries(1);
+  auto mock_catalog = std::make_shared<::testing::NiceMock<MockCatalog>>();
+  int update_calls = 0;
+  ON_CALL(*mock_catalog, UpdateTable(::testing::_, ::testing::_, ::testing::_))
+      .WillByDefault(
+          [&update_calls](const TableIdentifier&,
+                          const std::vector<std::unique_ptr<TableRequirement>>&,
+                          const std::vector<std::unique_ptr<TableUpdate>>&)
+              -> Result<std::shared_ptr<Table>> {
+            ++update_calls;
+            return CommitStateUnknown("injected unknown state");
+          });
+  ICEBERG_UNWRAP_OR_FAIL(auto mock_table,
+                         Table::Make(table_->name(), table_->metadata(),
+                                     std::string(table_->metadata_file_location()),
+                                     table_->io(), mock_catalog));
+
+  ICEBERG_UNWRAP_OR_FAIL(auto update, PositionDeleteUpdate::Make(mock_table));
+  update->Delete(data_file_->file_path, 6);
+  EXPECT_THAT(update->Commit(), IsError(ErrorKind::kCommitStateUnknown));
+  ASSERT_EQ(PuffinFiles().size(), 1);
+
+  EXPECT_THAT(update->Commit(), IsError(ErrorKind::kInvalidArgument));
+  EXPECT_EQ(update_calls, 1);
+  EXPECT_EQ(PuffinFiles().size(), 1);
+}
+
+}  // namespace
+
+}  // namespace iceberg