fix: enforce deletion vector scan applicability

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)
+	}
+}

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;
 }

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)

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) {