(db/snapshot): add CompoundSegmentsOperation and related test (#3057)

* (db/snapshot): update for row count
Signed-off-by: Npeng.xu <peng.xu@zilliz.com>

* (db/snapshot): fix bug in NewSegmentOperation
Signed-off-by: Npeng.xu <peng.xu@zilliz.com>

* (db/snapshot): remove dummy print
Signed-off-by: Npeng.xu <peng.xu@zilliz.com>

* (db/snapshot): Add some test for row count
Signed-off-by: Npeng.xu <peng.xu@zilliz.com>

* (db/snapshot): update size logic
Signed-off-by: Npeng.xu <peng.xu@zilliz.com>

* (db/snapshot): update size logic related ut
Signed-off-by: Npeng.xu <peng.xu@zilliz.com>

* (db/snapshot): rollback if operation is not done
Signed-off-by: Npeng.xu <peng.xu@zilliz.com>

* (db/snapshot): clean store
Signed-off-by: Npeng.xu <peng.xu@zilliz.com>

* (db/snapshot): remove some dependency
Signed-off-by: Npeng.xu <peng.xu@zilliz.com>

* (db/snapshot): update for store
Signed-off-by: Npeng.xu <peng.xu@zilliz.com>

* (db/snapshot): update Store.h
Signed-off-by: Npeng.xu <peng.xu@zilliz.com>

* (db/snapshot): update store related code
Signed-off-by: Npeng.xu <peng.xu@zilliz.com>

* (db/snapshot): add field element modification operation
Signed-off-by: Npeng.xu <peng.xu@zilliz.com>

* (db/snapshot): change new operation name
Signed-off-by: Npeng.xu <peng.xu@zilliz.com>

* (db/snapshot): fix lint error
Signed-off-by: Npeng.xu <peng.xu@zilliz.com>

* (db/snapshot): Add Segment File Operation
Signed-off-by: Npeng.xu <peng.xu@zilliz.com>

* (db/snapshot): crtp for BaseResource
Signed-off-by: Npeng.xu <peng.xu@zilliz.com>

* (db/snapshot): add InActiveResourcesGCEvent
Signed-off-by: Npeng.xu <peng.xu@zilliz.com>

* (db/snapshot): fix ut error
Signed-off-by: Npeng.xu <peng.xu@zilliz.com>

* (db/snapshot): small change
Signed-off-by: Npeng.xu <peng.xu@zilliz.com>

* (db/snapshot): update snapshot segmentcommit operation
Signed-off-by: Npeng.xu <peng.xu@zilliz.com>

* (db/snapshot): update drop all index operation
Signed-off-by: Npeng.xu <peng.xu@zilliz.com>

* (db/snapshot): update ut
Signed-off-by: Npeng.xu <peng.xu@zilliz.com>

* (db/snapshot): fix lint error
Signed-off-by: Npeng.xu <peng.xu@zilliz.com>

* (db/snapshot): fix gc
Signed-off-by: Npeng.xu <peng.xu@zilliz.com>

* (db/snapshot): fix gc segment files
Signed-off-by: Npeng.xu <peng.xu@zilliz.com>

* (db/snapshot): refactor Event related
Signed-off-by: Npeng.xu <peng.xu@zilliz.com>

* (db/snapshot): change 1
Signed-off-by: Npeng.xu <peng.xu@zilliz.com>

* (db/snapshot): change for GC event
Signed-off-by: Npeng.xu <peng.xu@zilliz.com>

* (db/snapshot): fix build error for high version of boost filesystem
Signed-off-by: Npeng.xu <peng.xu@zilliz.com>

* (db/snapshot): small change
Signed-off-by: Npeng.xu <peng.xu@zilliz.com>

* (db/snapshot): update compound operations
Signed-off-by: Npeng.xu <peng.xu@zilliz.com>

* (db/snapshot): add operation test
Signed-off-by: Npeng.xu <peng.xu@zilliz.com>

* (db/snapshot): update for CompoundSegmentsOperation
Signed-off-by: Npeng.xu <peng.xu@zilliz.com>

* (db/snapshot): add test for CompoundSegmentsOperation
Signed-off-by: Npeng.xu <peng.xu@zilliz.com>

core/src/db/snapshot/CompoundOperations.cpp

@@ -26,6 +26,158 @@ namespace milvus {
 namespace engine {
 namespace snapshot {
 
+CompoundSegmentsOperation::CompoundSegmentsOperation(const OperationContext& context, ScopedSnapshotT prev_ss)
+    : BaseT(context, prev_ss) {
+    for (auto& stale_segment_file : context_.stale_segment_files) {
+        stale_segment_files_[stale_segment_file->GetSegmentId()].push_back(stale_segment_file);
+        modified_segments_.insert(stale_segment_file->GetSegmentId());
+    }
+}
+
+Status
+CompoundSegmentsOperation::CommitRowCountDelta(ID_TYPE segment_id, SIZE_TYPE delta, bool sub) {
+    if (context_.new_segment && (context_.new_segment->GetID() == segment_id)) {
+        delta_[segment_id] = {delta, sub};
+    } else if (modified_segments_.find(segment_id) != modified_segments_.end()) {
+        delta_[segment_id] = {delta, sub};
+    } else {
+        return Status(SS_ERROR, "Cannot commit row count delta for segment " + std::to_string(segment_id));
+    }
+    return Status::OK();
+}
+
+Status
+CompoundSegmentsOperation::CommitNewSegment(const OperationContext& context, SegmentPtr& created) {
+    if (context_.new_segment) {
+        return Status(SS_DUPLICATED_ERROR, "Only one new segment could be created");
+    }
+    auto op = std::make_shared<SegmentOperation>(context, GetStartedSS());
+    STATUS_CHECK(op->Push());
+    STATUS_CHECK(op->GetResource(context_.new_segment));
+    created = context_.new_segment;
+    auto s_ctx_p = ResourceContextBuilder<Segment>().SetOp(meta::oUpdate).CreatePtr();
+    AddStepWithLsn(*created, context_.lsn, s_ctx_p);
+    return Status::OK();
+}
+
+Status
+CompoundSegmentsOperation::CommitNewSegmentFile(const SegmentFileContext& context, SegmentFilePtr& created) {
+    auto segment = GetStartedSS()->GetResource<Segment>(context.segment_id);
+    if (!segment) {
+        segment = context_.new_segment;
+    }
+
+    if (!segment || segment->GetID() != context.segment_id) {
+        std::stringstream emsg;
+        emsg << GetRepr() << ". Invalid segment " << context.segment_id << " in context";
+        return Status(SS_INVALID_CONTEX_ERROR, emsg.str());
+    }
+
+    auto ctx = context;
+    ctx.partition_id = segment->GetPartitionId();
+    auto new_sf_op = std::make_shared<SegmentFileOperation>(ctx, GetStartedSS());
+    STATUS_CHECK(new_sf_op->Push());
+    STATUS_CHECK(new_sf_op->GetResource(created));
+    new_segment_files_[created->GetSegmentId()].push_back(created);
+    modified_segments_.insert(created->GetSegmentId());
+    auto sf_ctx_p = ResourceContextBuilder<SegmentFile>().SetOp(meta::oUpdate).CreatePtr();
+    AddStepWithLsn(*created, context_.lsn, sf_ctx_p);
+
+    return Status::OK();
+}
+
+Status
+CompoundSegmentsOperation::DoExecute(StorePtr store) {
+    if (!context_.new_segment && stale_segment_files_.size() == 0 && new_segment_files_.size() == 0) {
+        return Status(SS_INVALID_CONTEX_ERROR, "Nothing to do");
+    }
+    if (context_.new_segment && context_.new_segment->IsActive()) {
+        return Status(SS_INVALID_CONTEX_ERROR, "New segment should not be active");
+    }
+
+    auto update_size = [&](SegmentFilePtr& file) {
+        auto update_ctx = ResourceContextBuilder<SegmentFile>().SetOp(meta::oUpdate).CreatePtr();
+        update_ctx->AddAttr(SizeField::Name);
+        AddStepWithLsn(*file, context_.lsn, update_ctx);
+    };
+
+    for (auto& kv : new_segment_files_) {
+        for (auto& new_file : kv.second) {
+            update_size(new_file);
+        }
+    }
+
+    if (context_.new_segment) {
+        modified_segments_.insert(context_.new_segment->GetID());
+    }
+
+    std::map<ID_TYPE, SegmentCommit::VecT> new_sc_map;
+    for (auto& m_seg_id : modified_segments_) {
+        OperationContext context;
+        context.lsn = context_.lsn;
+        auto itstale = stale_segment_files_.find(m_seg_id);
+        if (itstale != stale_segment_files_.end()) {
+            context.stale_segment_files = std::move(itstale->second);
+            stale_segment_files_.erase(itstale);
+        }
+        auto itnew = new_segment_files_.find(m_seg_id);
+        if (itnew != new_segment_files_.end()) {
+            context.new_segment_files = std::move(itnew->second);
+            new_segment_files_.erase(itnew);
+        }
+
+        if (context_.new_segment && context_.new_segment->GetID() == m_seg_id) {
+            context.new_segment = context_.new_segment;
+        }
+
+        SegmentCommitOperation sc_op(context, GetAdjustedSS());
+        STATUS_CHECK(sc_op(store));
+        SegmentCommitPtr new_sc;
+        STATUS_CHECK(sc_op.GetResource(new_sc));
+        auto segc_ctx_p = ResourceContextBuilder<SegmentCommit>().SetOp(meta::oUpdate).CreatePtr();
+        auto it_delta = delta_.find(m_seg_id);
+        if (it_delta != delta_.end()) {
+            auto delta = std::get<0>(it_delta->second);
+            auto is_sub = std::get<1>(it_delta->second);
+            if (delta != 0) {
+                auto new_row_cnt = 0;
+                if (is_sub && new_sc->GetRowCount() < delta) {
+                    return Status(SS_ERROR, "Invalid row count delta for segment " + std::to_string(m_seg_id));
+                } else if (is_sub) {
+                    new_row_cnt = new_sc->GetRowCount() - delta;
+                } else {
+                    new_row_cnt = new_sc->GetRowCount() + delta;
+                }
+                new_sc->SetRowCount(new_row_cnt);
+                segc_ctx_p->AddAttr(RowCountField::Name);
+            }
+        }
+
+        AddStepWithLsn(*new_sc, context.lsn, segc_ctx_p);
+        new_sc_map[new_sc->GetPartitionId()].push_back(new_sc);
+    }
+
+    for (auto& kv : new_sc_map) {
+        auto& partition_id = kv.first;
+        auto context = context_;
+        context.new_segment_commits = kv.second;
+        PartitionCommitOperation pc_op(context, GetAdjustedSS());
+        STATUS_CHECK(pc_op(store));
+        STATUS_CHECK(pc_op.GetResource(context.new_partition_commit));
+        auto pc_ctx_p = ResourceContextBuilder<PartitionCommit>().SetOp(meta::oUpdate).CreatePtr();
+        AddStepWithLsn(*context.new_partition_commit, context.lsn, pc_ctx_p);
+        context_.new_partition_commits.push_back(context.new_partition_commit);
+    }
+
+    CollectionCommitOperation cc_op(context_, GetAdjustedSS());
+    STATUS_CHECK(cc_op(store));
+    STATUS_CHECK(cc_op.GetResource(context_.new_collection_commit));
+    auto cc_ctx_p = ResourceContextBuilder<CollectionCommit>().SetOp(meta::oUpdate).CreatePtr();
+    AddStepWithLsn(*context_.new_collection_commit, context_.lsn, cc_ctx_p);
+
+    return Status::OK();
+}
+
 ChangeSegmentFileOperation::ChangeSegmentFileOperation(const OperationContext& context, ScopedSnapshotT prev_ss)
     : BaseT(context, prev_ss) {
 }

core/src/db/snapshot/CompoundOperations.h

@@ -11,7 +11,10 @@
 
 #pragma once
 
+#include <map>
+#include <set>
 #include <string>
+#include <utility>
 #include "ResourceOperations.h"
 #include "Snapshot.h"
 
@@ -61,7 +64,7 @@ class CompoundBaseOperation : public Operations {
 class ChangeSegmentFileOperation : public CompoundBaseOperation<ChangeSegmentFileOperation> {
  public:
     using BaseT = CompoundBaseOperation<ChangeSegmentFileOperation>;
-    static constexpr const char* Name = "B";
+    static constexpr const char* Name = "CSF";
 
     ChangeSegmentFileOperation(const OperationContext& context, ScopedSnapshotT prev_ss);
 
@@ -81,6 +84,31 @@ class ChangeSegmentFileOperation : public CompoundBaseOperation<ChangeSegmentFil
     bool sub_;
 };
 
+class CompoundSegmentsOperation : public CompoundBaseOperation<CompoundSegmentsOperation> {
+ public:
+    using BaseT = CompoundBaseOperation<CompoundSegmentsOperation>;
+    static constexpr const char* Name = "CS";
+
+    CompoundSegmentsOperation(const OperationContext& context, ScopedSnapshotT prev_ss);
+
+    Status DoExecute(StorePtr) override;
+
+    Status
+    CommitNewSegment(const OperationContext& context, SegmentPtr&);
+
+    Status
+    CommitNewSegmentFile(const SegmentFileContext& context, SegmentFilePtr& created);
+
+    Status
+    CommitRowCountDelta(ID_TYPE segment_id, SIZE_TYPE delta, bool sub = true);
+
+ protected:
+    std::map<ID_TYPE, std::pair<SIZE_TYPE, bool>> delta_;
+    std::map<ID_TYPE, SegmentFile::VecT> stale_segment_files_;
+    std::map<ID_TYPE, SegmentFile::VecT> new_segment_files_;
+    std::set<ID_TYPE> modified_segments_;
+};
+
 class AddFieldElementOperation : public CompoundBaseOperation<AddFieldElementOperation> {
  public:
     using BaseT = CompoundBaseOperation<AddFieldElementOperation>;

core/src/db/snapshot/ResourceOperations.cpp

@@ -203,6 +203,9 @@ SegmentCommitOperation::SegmentCommitOperation(const OperationContext& context,
 
 SegmentCommit::Ptr
 SegmentCommitOperation::GetPrevResource() const {
+    if (context_.new_segment) {
+        return nullptr;
+    }
     if (context_.new_segment_files.size() > 0) {
         return GetStartedSS()->GetSegmentCommitBySegmentId(context_.new_segment_files[0]->GetSegmentId());
     } else if (context_.stale_segment_files.size() > 0) {
@@ -225,9 +228,9 @@ SegmentCommitOperation::DoExecute(StorePtr store) {
             size -= stale_segment_file->GetSize();
         }
     } else {
-        resource_ = std::make_shared<SegmentCommit>(GetStartedSS()->GetLatestSchemaCommitId(),
-                                                    context_.new_segment_files[0]->GetPartitionId(),
-                                                    context_.new_segment_files[0]->GetSegmentId());
+        resource_ =
+            std::make_shared<SegmentCommit>(GetStartedSS()->GetLatestSchemaCommitId(),
+                                            context_.new_segment->GetPartitionId(), context_.new_segment->GetID());
     }
     for (auto& new_segment_file : context_.new_segment_files) {
         resource_->GetMappings().insert(new_segment_file->GetID());
@@ -240,9 +243,9 @@ SegmentCommitOperation::DoExecute(StorePtr store) {
 
 Status
 SegmentCommitOperation::PreCheck() {
-    if (context_.stale_segment_files.size() == 0 && context_.new_segment_files.size() == 0) {
+    if (context_.stale_segment_files.size() == 0 && context_.new_segment_files.size() == 0 && !context_.new_segment) {
         std::stringstream emsg;
-        emsg << GetRepr() << ". new_segment_files should not be empty in context";
+        emsg << GetRepr() << ". Segment files should not be empty in context";
         return Status(SS_INVALID_CONTEX_ERROR, emsg.str());
     }
     return Status::OK();

core/unittest/db/test_snapshot.cpp

@@ -938,7 +938,7 @@ TEST_F(SnapshotTest, OperationTest) {
     Snapshots::GetInstance().Reset();
 }
 
-TEST_F(SnapshotTest, ChangeSegmentFileOperationTest) {
+TEST_F(SnapshotTest, OperationTest2) {
     LSN_TYPE lsn = 0;
     std::string collection_name("c1");
     auto ss = CreateCollection(collection_name, ++lsn);
@@ -971,42 +971,181 @@ TEST_F(SnapshotTest, ChangeSegmentFileOperationTest) {
     auto stale_sf = ss->GetResource<SegmentFile>(*(sf_ids.begin()));
     ASSERT_TRUE(stale_sf);
 
-    std::cout << stale_sf->GetSize() << std::endl;
+    {
+        OperationContext context;
+        context.lsn = ++lsn;
+        context.stale_segment_files.push_back(stale_sf);
+        auto op = std::make_shared<ChangeSegmentFileOperation>(context, ss);
+        SegmentFilePtr seg_file;
+        sf_context.field_name = "vector";
+        sf_context.field_element_name = "_raw";
+        sf_context.segment_id = stale_sf->GetSegmentId();
+        sf_context.partition_id = stale_sf->GetPartitionId();
+        sf_context.collection_id = stale_sf->GetCollectionId();
+        status = op->CommitNewSegmentFile(sf_context, seg_file);
+        /* std::cout << status.ToString() << std::endl; */
+        ASSERT_TRUE(status.ok());
+        ASSERT_TRUE(seg_file);
 
-    OperationContext context;
-    context.lsn = ++lsn;
-    context.stale_segment_files.push_back(stale_sf);
-    auto op = std::make_shared<ChangeSegmentFileOperation>(context, ss);
-    SegmentFilePtr seg_file;
-    sf_context.field_name = "vector";
-    sf_context.field_element_name = "_raw";
-    sf_context.segment_id = stale_sf->GetSegmentId();
-    sf_context.partition_id = stale_sf->GetPartitionId();
-    sf_context.collection_id = stale_sf->GetCollectionId();
-    status = op->CommitNewSegmentFile(sf_context, seg_file);
-    /* std::cout << status.ToString() << std::endl; */
-    ASSERT_TRUE(status.ok());
-    ASSERT_TRUE(seg_file);
+        auto prev_segment_commit = ss->GetSegmentCommitBySegmentId(seg_file->GetSegmentId());
+        auto prev_segment_commit_mappings = prev_segment_commit->GetMappings();
+        ASSERT_FALSE(prev_segment_commit->ToString().empty());
 
-    auto prev_segment_commit = ss->GetSegmentCommitBySegmentId(seg_file->GetSegmentId());
-    auto prev_segment_commit_mappings = prev_segment_commit->GetMappings();
-    ASSERT_FALSE(prev_segment_commit->ToString().empty());
+        auto new_size = RandomInt(1000, 20000);
+        seg_file->SetSize(new_size);
+        total_size += new_size;
+        total_size -= stale_sf->GetSize();
 
-    auto new_size = RandomInt(1000, 20000);
-    seg_file->SetSize(new_size);
-    total_size += new_size;
-    total_size -= stale_sf->GetSize();
+        auto delta = prev_segment_commit->GetRowCount() / 2;
+        op->CommitRowCountDelta(delta);
+        total_row_cnt -= delta;
 
-    auto delta = prev_segment_commit->GetRowCount() / 2;
-    op->CommitRowCountDelta(delta);
-    total_row_cnt -= delta;
+        status = op->Push();
+        ASSERT_TRUE(status.ok()) << status.message();
+        status = op->GetSnapshot(ss);
+        ASSERT_TRUE(status.ok());
+        ASSERT_EQ(ss->GetCollectionCommit()->GetRowCount(), total_row_cnt);
+        ASSERT_EQ(ss->GetCollectionCommit()->GetSize(), total_size);
+        std::cout << ss->ToString() << std::endl;
+    }
 
-    status = op->Push();
-    ASSERT_TRUE(status.ok()) << status.message();
-    status = op->GetSnapshot(ss);
-    ASSERT_TRUE(status.ok());
-    ASSERT_EQ(ss->GetCollectionCommit()->GetRowCount(), total_row_cnt);
-    ASSERT_EQ(ss->GetCollectionCommit()->GetSize(), total_size);
+    {
+        OperationContext context;
+        context.lsn = ++lsn;
+
+        SegmentFile::VecT stale_sfs;
+        std::set<ID_TYPE> modified_segments;
+        auto executor = [&](const SegmentFilePtr& sf,
+                            SegmentFileIterator* itr) -> Status {
+            context.stale_segment_files.push_back(sf);
+            modified_segments.insert(sf->GetSegmentId());
+            return Status::OK();
+        };
+        auto iterator = std::make_shared<SegmentFileIterator>(ss, executor);
+        iterator->Iterate();
+        ASSERT_TRUE(iterator->GetStatus().ok());
+
+        {
+            auto op = std::make_shared<CompoundSegmentsOperation>(context, ss);
+
+            for (auto seg_id : modified_segments) {
+                auto sc = ss->GetSegmentCommitBySegmentId(seg_id);
+                ASSERT_TRUE(sc);
+                status = op->CommitRowCountDelta(seg_id, sc->GetRowCount() + 1, true);
+                ASSERT_TRUE(status.ok()) << status.ToString();
+            }
+
+            status = op->Push();
+            ASSERT_FALSE(status.ok()) << status.ToString();
+        }
+
+        {
+            auto op = std::make_shared<CompoundSegmentsOperation>(context, ss);
+            status = op->CommitRowCountDelta(100000000, 1, false);
+            ASSERT_FALSE(status.ok()) << status.ToString();
+        }
+
+        {
+            auto segment_cnt = ss->GetResources<Segment>().size();
+            context.lsn++;
+            auto op = std::make_shared<CompoundSegmentsOperation>(context, ss);
+            for (auto seg_id : modified_segments) {
+                auto sc = ss->GetSegmentCommitBySegmentId(seg_id);
+                ASSERT_TRUE(sc);
+                bool is_sub = (RandomInt(1, 10) >= 5);
+                auto delta = sc->GetRowCount() / 2;
+                status = op->CommitRowCountDelta(seg_id, delta, is_sub);
+                ASSERT_TRUE(status.ok()) << status.ToString();
+                if (is_sub) {
+                    total_row_cnt -= delta;
+                } else {
+                    total_row_cnt += delta;
+                }
+            }
+
+            auto new_size = 0;
+            for (auto stale_sf : context.stale_segment_files) {
+                SegmentFilePtr seg_file;
+                sf_context.field_name = "vector";
+                sf_context.field_element_name = "_raw";
+                sf_context.segment_id = stale_sf->GetSegmentId();
+                sf_context.partition_id = stale_sf->GetPartitionId();
+                sf_context.collection_id = stale_sf->GetCollectionId();
+                status = op->CommitNewSegmentFile(sf_context, seg_file);
+                auto size_delta = RandomInt(100, 1000);
+                seg_file->SetSize(size_delta);
+                new_size += size_delta;
+                ASSERT_TRUE(status.ok()) << status.ToString();
+            }
+
+            SegmentPtr new_seg;
+            OperationContext new_seg_ctx;
+            new_seg_ctx.prev_partition = ss->GetResource<Partition>(
+                    context.stale_segment_files[0]->GetPartitionId());
+            status = op->CommitNewSegment(new_seg_ctx, new_seg);
+            ASSERT_TRUE(status.ok()) << status.ToString();
+            ASSERT_TRUE(new_seg);
+
+            {
+                SegmentFilePtr seg_file;
+                sf_context.field_name = "vector";
+                sf_context.field_element_name = "_raw";
+                sf_context.segment_id = new_seg->GetID();
+                sf_context.partition_id = new_seg->GetPartitionId();
+                sf_context.collection_id = new_seg->GetCollectionId();
+                status = op->CommitNewSegmentFile(sf_context, seg_file);
+                ASSERT_TRUE(status.ok()) << status.ToString();
+                auto size_delta = RandomInt(100, 1000);
+                seg_file->SetSize(size_delta);
+                new_size += size_delta;
+                ASSERT_TRUE(status.ok()) << status.ToString();
+            }
+
+            auto delta_row = RandomInt(100, 1000);
+            status = op->CommitRowCountDelta(new_seg->GetID(), delta_row, false);
+            ASSERT_TRUE(status.ok());
+            total_row_cnt += delta_row;
+
+            status = op->Push();
+            ASSERT_TRUE(status.ok()) << status.ToString();
+            status = op->GetSnapshot(ss);
+            ASSERT_TRUE(status.ok());
+            std::cout << ss->ToString() << std::endl;
+            ASSERT_EQ(segment_cnt + 1, ss->GetResources<Segment>().size());
+
+            auto expect_size = ss->GetCollectionCommit()->GetSize();
+            ASSERT_EQ(expect_size, new_size) << status.ToString();
+            auto expect_row_cnt = ss->GetCollectionCommit()->GetRowCount();
+            ASSERT_EQ(expect_row_cnt, total_row_cnt) << status.ToString();
+        }
+
+        {
+            modified_segments.clear();
+            context.stale_segment_files.clear();
+            iterator = std::make_shared<SegmentFileIterator>(ss, executor);
+            iterator->Iterate();
+            ASSERT_TRUE(iterator->GetStatus().ok());
+
+            context.lsn++;
+            auto op = std::make_shared<CompoundSegmentsOperation>(context, ss);
+
+            for (auto seg_id : modified_segments) {
+                auto sc = ss->GetSegmentCommitBySegmentId(seg_id);
+                ASSERT_TRUE(sc);
+                status = op->CommitRowCountDelta(seg_id, sc->GetRowCount(), true);
+                ASSERT_TRUE(status.ok()) << status.ToString();
+            }
+
+            status = op->Push();
+            ASSERT_TRUE(status.ok()) << status.ToString();
+
+            status = op->GetSnapshot(ss);
+            ASSERT_TRUE(status.ok());
+            std::cout << ss->ToString() << std::endl;
+            ASSERT_EQ(ss->GetCollectionCommit()->GetRowCount(), 0);
+            ASSERT_EQ(ss->GetCollectionCommit()->GetSize(), 0);
+        }
+    }
 }
 
 TEST_F(SnapshotTest, CompoundTest1) {

core/unittest/db/utils.h

@@ -47,6 +47,7 @@ using DropIndexOperation = milvus::engine::snapshot::DropIndexOperation;
 using AddFieldElementOperation = milvus::engine::snapshot::AddFieldElementOperation;
 using DropAllIndexOperation = milvus::engine::snapshot::DropAllIndexOperation;
 using ChangeSegmentFileOperation = milvus::engine::snapshot::ChangeSegmentFileOperation;
+using CompoundSegmentsOperation = milvus::engine::snapshot::CompoundSegmentsOperation;
 using MergeOperation = milvus::engine::snapshot::MergeOperation;
 using CreateCollectionOperation = milvus::engine::snapshot::CreateCollectionOperation;
 using NewSegmentOperation = milvus::engine::snapshot::NewSegmentOperation;
@@ -64,6 +65,8 @@ using Segment = milvus::engine::snapshot::Segment;
 using SegmentPtr = milvus::engine::snapshot::SegmentPtr;
 using SegmentFile = milvus::engine::snapshot::SegmentFile;
 using SegmentFilePtr = milvus::engine::snapshot::SegmentFilePtr;
+using SegmentCommit = milvus::engine::snapshot::SegmentCommit;
+using SegmentCommitPtr = milvus::engine::snapshot::SegmentCommitPtr;
 using Field = milvus::engine::snapshot::Field;
 using FieldElement = milvus::engine::snapshot::FieldElement;
 using FieldElementPtr = milvus::engine::snapshot::FieldElementPtr;