fix: validate deletion vectors against Puffin metadata

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

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

src/iceberg/test/CMakeLists.txt

@@ -158,6 +158,8 @@ 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
                  puffin_reader_writer_test.cc)