fix: support all non-negative int64 deletion vector positions

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_writer.cc

@@ -19,7 +19,9 @@
 
 #include "iceberg/deletes/dv_writer.h"
 
+#include <cstddef>
 #include <cstdint>
+#include <limits>
 #include <map>
 #include <memory>
 #include <optional>
@@ -30,8 +32,8 @@
 #include <vector>
 
 #include "iceberg/deletes/dv_util_internal.h"
+#include "iceberg/deletes/dv_writer_internal.h"
 #include "iceberg/deletes/position_delete_index.h"
-#include "iceberg/deletes/roaring_position_bitmap.h"
 #include "iceberg/file_format.h"
 #include "iceberg/file_io.h"  // IWYU pragma: keep
 #include "iceberg/manifest/manifest_entry.h"
@@ -49,7 +51,8 @@ namespace iceberg {
 
 class DVWriter::Impl {
  public:
-  explicit Impl(DVWriterOptions options) : options_(std::move(options)) {}
+  Impl(DVWriterOptions options, size_t max_serialized_length)
+      : options_(std::move(options)), max_serialized_length_(max_serialized_length) {}
 
   // Accumulated positions and metadata for a single referenced data file.
   struct Deletes {
@@ -79,9 +82,7 @@ class DVWriter::Impl {
     ICEBERG_PRECHECK(!referenced_data_file.empty(),
                      "Deletion vector requires a non-empty referenced data file");
     ICEBERG_PRECHECK(spec != nullptr, "Deletion vector requires a partition spec");
-    ICEBERG_PRECHECK(pos >= 0 && pos <= RoaringPositionBitmap::kMaxPosition,
-                     "Deletion vector position out of range [0, {}]: {}",
-                     RoaringPositionBitmap::kMaxPosition, pos);
+    ICEBERG_PRECHECK(pos >= 0, "Deletion vector position must be non-negative: {}", pos);
     DeletesFor(referenced_data_file, spec, partition).positions.Delete(pos);
     return {};
   }
@@ -112,6 +113,11 @@ class DVWriter::Impl {
       ICEBERG_RETURN_UNEXPECTED(LoadPreviousDeletes(path, deletes));
     }
 
+    for (auto& [_, deletes] : deletes_by_path_) {
+      ICEBERG_RETURN_UNEXPECTED(
+          deletes.positions.ValidateSerializedSize(max_serialized_length_));
+    }
+
     ICEBERG_ASSIGN_OR_RAISE(auto output_file, options_.io->NewOutputFile(options_.path));
     const std::string output_path(options_.path);
     ICEBERG_ASSIGN_OR_RAISE(
@@ -198,19 +204,26 @@ class DVWriter::Impl {
   std::map<std::string, puffin::BlobMetadata, StringLess> blobs_by_path_;
   DeleteWriteResult result_;
   bool closed_ = false;
+  size_t max_serialized_length_;
 };
 
 DVWriter::DVWriter(std::unique_ptr<Impl> impl) : impl_(std::move(impl)) {}
 
 DVWriter::~DVWriter() = default;
 
 Result<std::unique_ptr<DVWriter>> DVWriter::Make(DVWriterOptions options) {
+  return internal::DVWriterFactory::Make(
+      std::move(options), static_cast<size_t>(std::numeric_limits<int32_t>::max()));
+}
+
+Result<std::unique_ptr<DVWriter>> internal::DVWriterFactory::Make(
+    DVWriterOptions options, size_t max_serialized_length) {
   ICEBERG_PRECHECK(!options.path.empty(), "DVWriter requires an output path");
   ICEBERG_PRECHECK(options.io != nullptr, "DVWriter requires a FileIO");
   ICEBERG_PRECHECK(options.load_previous_deletes != nullptr,
                    "DVWriter requires a load_previous_deletes callback");
-  return std::unique_ptr<DVWriter>(
-      new DVWriter(std::make_unique<Impl>(std::move(options))));
+  return std::unique_ptr<DVWriter>(new DVWriter(
+      std::make_unique<DVWriter::Impl>(std::move(options), max_serialized_length)));
 }
 
 Status DVWriter::Delete(std::string_view referenced_data_file, int64_t pos,

src/iceberg/deletes/dv_writer.h

@@ -38,6 +38,10 @@
 
 namespace iceberg {
 
+namespace internal {
+class DVWriterFactory;
+}
+
 /// \brief File metadata for deletion vectors produced by DVWriter.
 struct ICEBERG_EXPORT DeleteWriteResult {
   /// Deletion vector files produced by the writer.
@@ -87,6 +91,8 @@ class ICEBERG_EXPORT DVWriter {
   std::unique_ptr<Impl> impl_;
 
   explicit DVWriter(std::unique_ptr<Impl> impl);
+
+  friend class internal::DVWriterFactory;
 };
 
 }  // namespace iceberg

src/iceberg/deletes/dv_writer_internal.h

@@ -0,0 +1,40 @@
+/*
+ * 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/deletes/dv_writer_internal.h
+/// Internal deletion vector writer helpers.
+
+#include <cstddef>
+#include <memory>
+
+#include "iceberg/deletes/dv_writer.h"
+#include "iceberg/iceberg_export.h"
+#include "iceberg/result.h"
+
+namespace iceberg::internal {
+
+class ICEBERG_EXPORT DVWriterFactory {
+ public:
+  static Result<std::unique_ptr<DVWriter>> Make(DVWriterOptions options,
+                                                size_t max_serialized_length);
+};
+
+}  // namespace iceberg::internal

src/iceberg/deletes/position_delete_index.cc

@@ -49,6 +49,7 @@ constexpr std::array<uint8_t, 4> kMagic = {0xD1, 0xD3, 0x39, 0x64};
 constexpr int32_t kLengthPrefixBytes = 4;
 constexpr int32_t kMagicBytes = 4;
 constexpr int32_t kCrcBytes = 4;
+constexpr size_t kMaxSerializedLength = std::numeric_limits<int32_t>::max();
 
 uint32_t ComputeCrc32(std::span<const uint8_t> bytes) {
   uLong crc = crc32(0L, Z_NULL, 0);
@@ -142,16 +143,24 @@ void PositionDeleteIndex::Merge(const PositionDeleteIndex& other) {
                        other.delete_files_.end());
 }
 
-Result<std::vector<uint8_t>> PositionDeleteIndex::Serialize() {
+Status PositionDeleteIndex::ValidateSerializedSize(size_t max_length) {
   bitmap_.Optimize();  // run-length encode before serializing
-  std::vector<uint8_t> blob(kLengthPrefixBytes);
-  blob.insert(blob.end(), kMagic.begin(), kMagic.end());
-  ICEBERG_ASSIGN_OR_RAISE(const auto vector_size, bitmap_.SerializeTo(blob));
-
+  const size_t vector_size = bitmap_.SerializedSizeInBytes();
   const size_t magic_and_vector_size = kMagicBytes + vector_size;
-  ICEBERG_PRECHECK(magic_and_vector_size <= std::numeric_limits<int32_t>::max(),
+  ICEBERG_PRECHECK(magic_and_vector_size <= max_length,
                    "Deletion vector is too large to serialize: {} bytes",
                    magic_and_vector_size);
+  return {};
+}
+
+Result<std::vector<uint8_t>> PositionDeleteIndex::Serialize() {
+  ICEBERG_RETURN_UNEXPECTED(ValidateSerializedSize(kMaxSerializedLength));
+  const size_t vector_size = bitmap_.SerializedSizeInBytes();
+  const size_t magic_and_vector_size = kMagicBytes + vector_size;
+
+  std::vector<uint8_t> blob(kLengthPrefixBytes);
+  blob.insert(blob.end(), kMagic.begin(), kMagic.end());
+  ICEBERG_RETURN_UNEXPECTED(bitmap_.SerializeTo(blob));
 
   WriteBigEndian(static_cast<int32_t>(magic_and_vector_size), blob.data());
   const auto crc_offset = blob.size();

src/iceberg/deletes/position_delete_index.h

@@ -22,6 +22,7 @@
 /// \file iceberg/deletes/position_delete_index.h
 /// Index of deleted row positions for a data file.
 
+#include <cstddef>
 #include <cstdint>
 #include <memory>
 #include <span>
@@ -34,6 +35,8 @@
 
 namespace iceberg {
 
+class DVWriter;
+
 /// \brief Tracks deleted row positions using a bitmap.
 ///
 /// This class provides a domain-specific API for position deletes
@@ -53,6 +56,8 @@ class ICEBERG_EXPORT PositionDeleteIndex {
   /// \brief Mark a range of positions as deleted [pos_start, pos_end).
   /// \param pos_start Start position (inclusive)
   /// \param pos_end End position (exclusive)
+  /// \note Because pos_end is an int64_t exclusive endpoint, this method cannot
+  ///       include INT64_MAX. Call Delete(INT64_MAX) separately.
   void Delete(int64_t pos_start, int64_t pos_end);
 
   /// \brief Check if a position is deleted.
@@ -97,6 +102,8 @@ class ICEBERG_EXPORT PositionDeleteIndex {
  private:
   explicit PositionDeleteIndex(RoaringPositionBitmap bitmap);
 
+  Status ValidateSerializedSize(size_t max_length);
+
   // Bulk-add positions sharing high-32-bit `key`. Private hook for
   // `ForEachPositionDelete`'s bulk path; keeps `Delete` the sole public
   // mutation surface.
@@ -105,6 +112,7 @@ class ICEBERG_EXPORT PositionDeleteIndex {
   friend void ICEBERG_EXPORT ForEachPositionDelete(std::span<const int64_t> positions,
                                                    PositionDeleteIndex& target,
                                                    std::vector<uint32_t>& scratch);
+  friend class DVWriter;
 
   RoaringPositionBitmap bitmap_;
   std::vector<std::shared_ptr<DataFile>> delete_files_;