feat: Add merge method for DeletionVector

src/paimon/core/deletionvectors/bitmap_deletion_vector.cpp

@@ -84,4 +84,18 @@ Result<PAIMON_UNIQUE_PTR<DeletionVector>> BitmapDeletionVector::Deserialize(cons
                                          length - MAGIC_NUMBER_SIZE_BYTES, pool);
 }
 
+Status BitmapDeletionVector::Merge(const std::shared_ptr<DeletionVector>& deletion_vector) {
+    if (!deletion_vector || deletion_vector->IsEmpty()) {
+        return Status::OK();
+    }
+    auto* other = dynamic_cast<BitmapDeletionVector*>(deletion_vector.get());
+    if (other != nullptr) {
+        roaring_bitmap_ |= other->roaring_bitmap_;
+    } else {
+        return Status::Invalid(
+            "Cannot merge a non-BitmapDeletionVector into a BitmapDeletionVector");
+    }
+    return Status::OK();
+}
+
 }  // namespace paimon

src/paimon/core/deletionvectors/bitmap_deletion_vector.h

@@ -64,6 +64,8 @@ class BitmapDeletionVector : public DeletionVector {
     Result<PAIMON_UNIQUE_PTR<Bytes>> SerializeToBytes(
         const std::shared_ptr<MemoryPool>& pool) override;
 
+    Status Merge(const std::shared_ptr<DeletionVector>& deletion_vector) override;
+
     const RoaringBitmap32* GetBitmap() const {
         return &roaring_bitmap_;
     }

src/paimon/core/deletionvectors/bitmap_deletion_vector_test.cpp

@@ -165,4 +165,73 @@ TEST(BitmapDeletionVectorTest, DeserializeWithoutMagicNumberShouldRoundTrip) {
     ASSERT_FALSE(deserialized->IsDeleted(8).value());
 }
 
+TEST(BitmapDeletionVectorTest, MergeTwoBitmapDeletionVectors) {
+    RoaringBitmap32 roaring1;
+    roaring1.Add(1);
+    roaring1.Add(3);
+    roaring1.Add(5);
+    auto dv1 = std::make_shared<BitmapDeletionVector>(roaring1);
+
+    RoaringBitmap32 roaring2;
+    roaring2.Add(2);
+    roaring2.Add(4);
+    roaring2.Add(5);  // overlapping position
+    auto dv2 = std::make_shared<BitmapDeletionVector>(roaring2);
+
+    ASSERT_OK(dv1->Merge(dv2));
+
+    // All positions from both vectors should be marked as deleted.
+    ASSERT_TRUE(dv1->IsDeleted(1).value());
+    ASSERT_TRUE(dv1->IsDeleted(2).value());
+    ASSERT_TRUE(dv1->IsDeleted(3).value());
+    ASSERT_TRUE(dv1->IsDeleted(4).value());
+    ASSERT_TRUE(dv1->IsDeleted(5).value());
+    ASSERT_FALSE(dv1->IsDeleted(0).value());
+    ASSERT_FALSE(dv1->IsDeleted(6).value());
+    ASSERT_EQ(dv1->GetCardinality(), 5);
+}
+
+TEST(BitmapDeletionVectorTest, MergeEmptyDeletionVector) {
+    RoaringBitmap32 roaring1;
+    roaring1.Add(10);
+    roaring1.Add(20);
+    auto dv1 = std::make_shared<BitmapDeletionVector>(roaring1);
+
+    RoaringBitmap32 empty_roaring;
+    auto dv_empty = std::make_shared<BitmapDeletionVector>(empty_roaring);
+
+    ASSERT_OK(dv1->Merge(dv_empty));
+    ASSERT_EQ(dv1->GetCardinality(), 2);
+    ASSERT_TRUE(dv1->IsDeleted(10).value());
+    ASSERT_TRUE(dv1->IsDeleted(20).value());
+}
+
+TEST(BitmapDeletionVectorTest, MergeNullDeletionVector) {
+    RoaringBitmap32 roaring1;
+    roaring1.Add(7);
+    auto dv1 = std::make_shared<BitmapDeletionVector>(roaring1);
+
+    ASSERT_OK(dv1->Merge(nullptr));
+    ASSERT_EQ(dv1->GetCardinality(), 1);
+    ASSERT_TRUE(dv1->IsDeleted(7).value());
+}
+
+TEST(BitmapDeletionVectorTest, MergeIntoEmptyDeletionVector) {
+    RoaringBitmap32 empty_roaring;
+    auto dv1 = std::make_shared<BitmapDeletionVector>(empty_roaring);
+    ASSERT_TRUE(dv1->IsEmpty());
+
+    RoaringBitmap32 roaring2;
+    roaring2.Add(100);
+    roaring2.Add(200);
+    roaring2.Add(300);
+    auto dv2 = std::make_shared<BitmapDeletionVector>(roaring2);
+
+    ASSERT_OK(dv1->Merge(dv2));
+    ASSERT_EQ(dv1->GetCardinality(), 3);
+    ASSERT_TRUE(dv1->IsDeleted(100).value());
+    ASSERT_TRUE(dv1->IsDeleted(200).value());
+    ASSERT_TRUE(dv1->IsDeleted(300).value());
+}
+
 }  // namespace paimon::test

src/paimon/core/deletionvectors/deletion_vector.h

@@ -79,6 +79,15 @@ class DeletionVector {
         return is_valid;
     }
 
+    /// Merges another DeletionVector into this current one.
+    ///
+    /// This method combines the deletion positions from the other deletion vector
+    /// with the current one, marking all positions deleted in either vector as deleted.
+    ///
+    /// @param deletion_vector The other DeletionVector to merge into this one.
+    /// @return Status indicating success or failure of the merge operation.
+    virtual Status Merge(const std::shared_ptr<DeletionVector>& deletion_vector) = 0;
+
     /// Determines if the deletion vector is empty, indicating no deletions.
     ///
     /// @return true if the deletion vector is empty, false if it contains deletions.

src/paimon/core/deletionvectors/deletion_vector_index_file_writer_test.cpp

@@ -60,6 +60,10 @@ class FailingDeletionVector : public DeletionVector {
     Result<PAIMON_UNIQUE_PTR<Bytes>> SerializeToBytes(const std::shared_ptr<MemoryPool>&) override {
         return Status::Invalid("injected serialize failure");
     }
+
+    Status Merge(const std::shared_ptr<DeletionVector>&) override {
+        return Status::Invalid("injected merge failure");
+    }
 };
 
 }  // namespace