milvus/tests/python_client/testcases/test_compaction.py

from time import monotonic, sleep, time

import pytest
from base.client_base import TestcaseBase
from common import common_func as cf
from common import common_type as ct
from common.common_type import CaseLabel, CheckTasks
from common.compaction_utils import run_compaction_with_workload
from pymilvus.exceptions import MilvusException
from utils.util_log import test_log as log

prefix = "compact"
tmp_nb = 100


def remaining_time(deadline, operation):
    remaining = deadline - monotonic()
    if remaining <= 0:
        raise TimeoutError(f"Timed out while {operation}")
    return remaining


def require_observed_overlap(observed):
    if not observed:
        pytest.skip("Insufficient concurrency coverage: no Executing sample during an active workload call")


class TestCompactionParams(TestcaseBase):
    @pytest.mark.tags(CaseLabel.L2)
    def test_compact_without_connection(self):
        """
        target: test compact without connection
        method: compact after remove connection
        expected: raise exception
        """
        # init collection with tmp_nb default data
        collection_w = self.init_collection_general(prefix, nb=tmp_nb, insert_data=True)[0]

        # remove connection and delete
        self.connection_wrap.remove_connection(ct.default_alias)
        res_list, _ = self.connection_wrap.list_connections()
        assert ct.default_alias not in res_list
        error = {ct.err_code: 999, ct.err_msg: "should create connection first"}
        collection_w.compact(check_task=CheckTasks.err_res, check_items=error)

    @pytest.mark.tags(CaseLabel.L1)
    @pytest.mark.skip(
        "DataCoord: for A, B -> C, will not compact segment C before A, "
        "B GCed, no method to check whether a segment is GCed"
    )
    def test_compact_twice(self):
        """
        target: test compact twice
        method: 1.create with shard_num=1
                2.insert and flush twice (two segments)
                3.compact
                4.insert new data
                5.compact
        expected: Merge into one segment
        """
        # init collection with one shard, insert into two segments
        pytest.skip(
            "DataCoord: for A, B -> C, will not compact segment C before A, B GCed, no method to check whether a segment is GCed"
        )
        collection_w = self.collection_insert_multi_segments_one_shard(prefix, nb_of_segment=tmp_nb)

        # first compact two segments
        collection_w.compact()
        collection_w.wait_for_compaction_completed()
        c_plans1 = collection_w.get_compaction_plans()[0]
        target_1 = c_plans1.plans[0].target

        # insert new data
        df = cf.gen_default_dataframe_data(tmp_nb)
        collection_w.insert(df)
        log.debug(collection_w.num_entities)

        # second compact
        collection_w.compact()
        collection_w.wait_for_compaction_completed()
        collection_w.get_compaction_state()
        c_plans2 = collection_w.get_compaction_plans()[0]

        assert target_1 in c_plans2.plans[0].sources
        log.debug(c_plans2.plans[0].target)

    @pytest.mark.tags(CaseLabel.L1)
    @pytest.mark.skip(reason="https://github.com/milvus-io/milvus/issues/20747")
    def test_compact_partition(self):
        """
        target: test compact partition
        method: compact partition
        expected: Verify partition segments merged
        """
        # create collection with shard_num=1, and create partition
        collection_w = self.init_collection_wrap(name=cf.gen_unique_str(prefix), shards_num=1)
        partition_w = self.init_partition_wrap(collection_wrap=collection_w)

        # insert flush twice
        for i in range(2):
            df = cf.gen_default_dataframe_data(tmp_nb)
            partition_w.insert(df)
            assert partition_w.num_entities == tmp_nb * (i + 1)

        # create index
        collection_w.create_index(ct.default_float_vec_field_name, ct.default_index)
        log.debug(collection_w.index())

        # compact
        collection_w.compact()
        collection_w.wait_for_compaction_completed()
        c_plans = collection_w.get_compaction_plans()[0]

        assert len(c_plans.plans) == 1
        assert len(c_plans.plans[0].sources) == 2
        target = c_plans.plans[0].target

        # verify queryNode load the compacted segments
        cost = 180
        start = time()
        while time() - start < cost:
            collection_w.load()
            segment_info = self.utility_wrap.get_query_segment_info(collection_w.name)[0]
            if len(segment_info) == 1:
                break
            sleep(1.0)
        assert target == segment_info[0].segmentID

    @pytest.mark.tags(CaseLabel.L2)
    def test_compact_only_growing_segment(self):
        """
        target: test compact growing data
        method: 1.insert into multi segments without flush
                2.compact
        expected: No compaction (compact just for sealed data)
        """
        # create and insert without flush
        collection_w = self.init_collection_wrap(name=cf.gen_unique_str(prefix))
        df = cf.gen_default_dataframe_data(tmp_nb)
        collection_w.insert(df)

        # compact when only growing segment
        collection_w.compact()
        collection_w.wait_for_compaction_completed()
        c_plans = collection_w.get_compaction_plans()[0]
        assert len(c_plans.plans) == 0

    @pytest.mark.tags(CaseLabel.L2)
    def test_compact_empty_collection(self):
        """
        target: test compact an empty collection
        method: compact an empty collection
        expected: No exception
        """
        # init collection and empty
        collection_w = self.init_collection_wrap(name=cf.gen_unique_str(prefix))

        # compact
        collection_w.compact()
        c_plans, _ = collection_w.get_compaction_plans()
        assert len(c_plans.plans) == 0

    @pytest.mark.tags(CaseLabel.L3)
    @pytest.mark.parametrize("delete_pos", [1, tmp_nb // 2])
    def test_compact_after_delete_partial(self, delete_pos):
        """
        target: test delete one entity and compact
        method: 1.create with shard_num=1
                2.delete one sealed entity, half entities
                2.compact
        expected: Verify compact result
        """
        # create, insert without flush
        collection_w = self.init_collection_wrap(cf.gen_unique_str(prefix), shards_num=1)
        df = cf.gen_default_dataframe_data(tmp_nb)
        insert_res, _ = collection_w.insert(df)

        # delete single entity, flush
        single_expr = f"{ct.default_int64_field_name} in {insert_res.primary_keys[:delete_pos]}"
        collection_w.delete(single_expr)
        collection_w.flush()

        # compact, get plan
        sleep(ct.compact_retention_duration + 1)
        collection_w.compact()
        collection_w.wait_for_compaction_completed()
        collection_w.get_compaction_plans(check_task=CheckTasks.check_delete_compact)

        collection_w.create_index(ct.default_float_vec_field_name, index_params=ct.default_flat_index)
        collection_w.load()
        collection_w.query(single_expr, check_task=CheckTasks.check_query_empty)

        res = df.iloc[-1:, :1].to_dict("records")
        collection_w.query(
            f"{ct.default_int64_field_name} in {insert_res.primary_keys[-1:]}",
            check_task=CheckTasks.check_query_results,
            check_items={"exp_res": res},
        )

    @pytest.mark.tags(CaseLabel.L3)
    def test_compact_after_delete_index(self):
        """
        target: test compact after delete and create index
        method: 1.create with 1 shard and insert nb entities (ensure can be index)
                2.delete some entities and flush (ensure generate delta log)
                3.create index
                4.compact outside retentionDuration
                5.load and search
        expected: Empty search result
        """
        # create, insert without flush
        collection_w = self.init_collection_wrap(cf.gen_unique_str(prefix), shards_num=1)
        df = cf.gen_default_dataframe_data()
        insert_res, _ = collection_w.insert(df)

        # delete and flush
        expr = f"{ct.default_int64_field_name} in {insert_res.primary_keys[: ct.default_nb // 2]}"
        collection_w.delete(expr)
        assert collection_w.num_entities == ct.default_nb

        # build index
        collection_w.create_index(ct.default_float_vec_field_name, ct.default_index)
        log.debug(collection_w.index())

        # compact, get plan
        sleep(ct.compact_retention_duration + 1)
        collection_w.compact()
        collection_w.wait_for_compaction_completed()
        collection_w.get_compaction_plans(check_task=CheckTasks.check_delete_compact)

        collection_w.load()
        res, _ = collection_w.search(
            df[ct.default_float_vec_field_name][:1].to_list(),
            ct.default_float_vec_field_name,
            ct.default_search_params,
            ct.default_limit,
        )
        assert len(res[0]) == ct.default_limit

    @pytest.mark.tags(CaseLabel.L2)
    @pytest.mark.skip("https://github.com/milvus-io/milvus/issues/31258")
    def test_compact_delete_ratio(self):
        """
        target: test delete entities reaches ratio and auto-compact
        method: 1.create with shard_num=1
                2.insert (compact load delta log, not from dmlChannel)
                3.delete 20% of nb, flush
        expected: Verify auto compaction, merge insert log and delta log
        """
        collection_w = self.init_collection_wrap(name=cf.gen_unique_str(prefix), shards_num=1)
        df = cf.gen_default_dataframe_data(tmp_nb)
        insert_res, _ = collection_w.insert(df)

        collection_w.create_index(ct.default_float_vec_field_name, index_params=ct.default_flat_index)
        # delete 20% entities
        ratio_expr = (
            f"{ct.default_int64_field_name} in {insert_res.primary_keys[: tmp_nb // ct.compact_delta_ratio_reciprocal]}"
        )
        collection_w.delete(ratio_expr)
        collection_w.flush()

        # Flush a new segment and meet condition 20% deleted entities, triggre compaction but no way to get plan
        collection_w.insert(cf.gen_default_dataframe_data(1, start=tmp_nb))

        exp_num_entities_after_compact = tmp_nb - (tmp_nb // ct.compact_delta_ratio_reciprocal) + 1
        start = time()
        while True:
            if collection_w.num_entities == exp_num_entities_after_compact:
                break
            if time() - start > 360:
                raise MilvusException(1, "Auto delete ratio compaction cost more than 360s")
            sleep(1)

        collection_w.load()
        collection_w.query(ratio_expr, check_task=CheckTasks.check_query_empty)

        res = df.iloc[-1:, :1].to_dict("records")
        collection_w.query(
            f"{ct.default_int64_field_name} in {insert_res.primary_keys[-1:]}",
            check_task=CheckTasks.check_query_results,
            check_items={"exp_res": res},
        )

    @pytest.mark.tags(CaseLabel.L2)
    def test_compact_delete_less_ratio(self):
        """
        target: test delete entities less ratio and no compact
        method: 1.create collection shard_num=1
                2.insert without flush
                3.delete 10% entities and flush
        expected: Verify no compact (can't), delete successfully
        """
        collection_w = self.init_collection_wrap(name=cf.gen_unique_str(prefix), shards_num=1)
        df = cf.gen_default_dataframe_data(tmp_nb)
        insert_res, _ = collection_w.insert(df)

        # delete 10% entities, ratio = 0.1
        less_ratio_reciprocal = 10
        ratio_expr = f"{ct.default_int64_field_name} in {insert_res.primary_keys[: tmp_nb // less_ratio_reciprocal]}"
        collection_w.delete(ratio_expr)
        assert collection_w.num_entities == tmp_nb

        collection_w.create_index(ct.default_float_vec_field_name, index_params=ct.default_flat_index)
        collection_w.load()
        collection_w.query(ratio_expr, check_task=CheckTasks.check_query_empty)

    @pytest.mark.tags(CaseLabel.L3)
    def test_compact_after_delete_all(self):
        """
        target: test delete all and compact
        method: 1.create with shard_num=1
                2.delete all sealed data
                3.compact
        expected: Get compaction plan
        """
        collection_w = self.init_collection_wrap(name=cf.gen_unique_str(prefix), shards_num=1)
        df = cf.gen_default_dataframe_data()
        res, _ = collection_w.insert(df)

        expr = f"{ct.default_int64_field_name} in {res.primary_keys}"
        collection_w.delete(expr)
        assert collection_w.num_entities == ct.default_nb

        # currently no way to verify whether it is compact after delete,
        # because the merge compact plan is generate first
        sleep(ct.compact_retention_duration + 1)
        collection_w.compact()
        collection_w.wait_for_compaction_completed()
        collection_w.get_compaction_plans(check_task=CheckTasks.check_delete_compact)

        collection_w.create_index(ct.default_float_vec_field_name, index_params=ct.default_flat_index)
        collection_w.load()
        collection_w.query(expr, check_task=CheckTasks.check_query_empty)

    @pytest.mark.tags(CaseLabel.L2)
    def test_compact_after_delete(self):
        """
        target: test delete and then compact
        method: 1. create a collection and insert data
                2. delete all data and compact
                3. load query and release
        expected: can't find the deleted data
        """
        nb = 50000
        collection_w = self.init_collection_wrap(name=cf.gen_unique_str(prefix))
        df = cf.gen_default_dataframe_data(nb=nb)
        res, _ = collection_w.insert(df)
        assert collection_w.num_entities == nb

        expr = f"{ct.default_int64_field_name} in {res.primary_keys}"
        collection_w.create_index(ct.default_float_vec_field_name, index_params=ct.default_flat_index)
        collection_w.delete(expr)
        assert collection_w.num_entities == nb
        collection_w.compact()
        collection_w.wait_for_compaction_completed()
        # collection_w.get_compaction_plans(check_task=CheckTasks.check_delete_compact)

        collection_w.load()
        res = collection_w.query(expr)[0]
        assert len(res) == 0
        collection_w.release()

        sleep(30)
        collection_w.load()
        res = collection_w.query(expr)[0]
        assert len(res) == 0
        collection_w.release()

        sleep(40)
        collection_w.load()
        res = collection_w.query(expr)[0]
        assert len(res) == 0
        collection_w.release()

        sleep(60)
        collection_w.load()
        res = collection_w.query(expr)[0]
        assert len(res) == 0
        collection_w.release()

    @pytest.mark.skip(reason="TODO")
    @pytest.mark.tags(CaseLabel.L2)
    def test_compact_delete_max_delete_size(self):
        """
        target: test compact delta log reaches max delete size 10MiB
        method: todo
        expected: auto merge single segment
        """
        pass

    @pytest.mark.tags(CaseLabel.L1)
    def test_compact_max_time_interval(self):
        """
        target: test auto compact with max interval 60s
        method: 1.create with shard_num=1
                2.insert flush twice (two segments)
                3.wait max_compaction_interval (60s)
        expected: Verify compaction results
        """
        # create collection shard_num=1, insert 2 segments, each with tmp_nb entities
        collection_w = self.init_collection_wrap(name=cf.gen_unique_str(prefix), shards_num=1)

        # Notice:The merge segments compaction triggered by max_compaction_interval also needs to meet
        # the compaction_segment_ num_threshold
        for i in range(ct.compact_segment_num_threshold):
            df = cf.gen_default_dataframe_data(tmp_nb)
            collection_w.insert(df)
            assert collection_w.num_entities == tmp_nb * (i + 1)

        collection_w.create_index(ct.default_float_vec_field_name, ct.default_index)
        collection_w.compact()
        sleep(10)  # modified by nico, debug compaction time from 60s to 10s

        # verify queryNode load the compacted segments
        collection_w.load()
        replicas = collection_w.get_replicas()[0]
        replica_num = len(replicas.groups)
        cost = 180
        start = time()
        while time() - start < cost:
            sleep(1.0)
            collection_w.load()
            segment_info = self.utility_wrap.get_query_segment_info(collection_w.name)[0]
            if len(segment_info) == 1 * replica_num:
                break
            if time() - start > cost:
                raise MilvusException(1, f"Waiting more than {cost}s for the compacted segment indexed")
            collection_w.release()

    @pytest.mark.skip(reason="TODO")
    @pytest.mark.tags(CaseLabel.L2)
    def test_compact_delta_max_time_interval(self):
        """
        target: test merge insert and delta log triggered by max_compaction_interval
        method: todo
        expected: auto compact binlogs
        """
        pass


class TestCompactionOperation(TestcaseBase):
    # @pytest.mark.xfail(reason="Issue https://github.com/milvus-io/milvus/issues/15665")
    @pytest.mark.tags(CaseLabel.L3)
    def test_compact_both_delete_merge(self):
        """
        target: test compact both delete and merge
        method: 1.create collection with shard_num=1
                2.insert data into two segments
                3.delete and flush (new insert)
                4.compact
                5.load and search
        expected: Triggre two types compaction
        """
        collection_w = self.init_collection_wrap(cf.gen_unique_str(prefix), shards_num=1)
        ids = []
        for i in range(2):
            df = cf.gen_default_dataframe_data(tmp_nb, start=i * tmp_nb)
            insert_res, _ = collection_w.insert(df)
            assert collection_w.num_entities == (i + 1) * tmp_nb
            ids.extend(insert_res.primary_keys)

        # delete_ids = ids[:tmp_nb]
        delete_ids = [0, tmp_nb // 2]
        expr = f"{ct.default_int64_field_name} in {delete_ids}"
        collection_w.delete(expr)

        collection_w.insert(cf.gen_default_dataframe_data(1, start=2 * tmp_nb))
        assert collection_w.num_entities == 2 * tmp_nb + 1

        sleep(ct.compact_retention_duration + 1)

        collection_w.compact()
        collection_w.wait_for_compaction_completed()
        collection_w.get_compaction_plans()

        # todo assert two types of compaction plan (delete + merge). The historical
        # `assert len(plans) == 2` was never enabled, so the expected count is unverified.

        # search
        ids.pop(0)
        ids.pop(-1)
        collection_w.load()
        search_res, _ = collection_w.search(
            cf.gen_vectors(ct.default_nq, ct.default_dim),
            ct.default_float_vec_field_name,
            ct.default_search_params,
            ct.default_limit,
            check_items={"nq": ct.default_nq, "ids": ids, "limit": ct.default_limit},
        )
        collection_w.query(f"{ct.default_int64_field_name} in {delete_ids}", check_task=CheckTasks.check_query_empty)

    @pytest.mark.tags(CaseLabel.L3)
    def test_compact_delete_multi_segments(self):
        """
        target: test compact multi delete segments
        method: 1.create collection with shard_num=2
                2.insert data into two segments
                3.delete entities from two segments
                4.compact
                5.load and search
        expected: Verify two delta compaction plans
        """
        collection_w = self.init_collection_wrap(cf.gen_unique_str(prefix))
        df = cf.gen_default_dataframe_data(2 * tmp_nb)
        insert_res, _ = collection_w.insert(df)
        assert collection_w.num_entities == 2 * tmp_nb

        collection_w.load()
        log.debug(self.utility_wrap.get_query_segment_info(collection_w.name))

        delete_ids = [i for i in range(10)]
        expr = f"{ct.default_int64_field_name} in {delete_ids}"
        collection_w.delete(expr)

        sleep(ct.compact_retention_duration + 1)

        collection_w.compact()
        collection_w.wait_for_compaction_completed()
        c_plans = collection_w.get_compaction_plans()[0]
        assert len(c_plans.plans) == 2
        for plan in c_plans.plans:
            assert len(plan.sources) == 1

        collection_w.query(f"{ct.default_int64_field_name} in {delete_ids}", check_task=CheckTasks.check_query_empty)

    @pytest.mark.tags(CaseLabel.L2)
    def test_compact_merge_multi_shards(self):
        """
        target: test compact merge multi shards
        method: 1.Create a collection with 2 shards
                2.Insert twice and generate 4 segments
                3.Compact and wait it completed
        expected: Verify there are 2 merge type complation plans
        """
        collection_w = self.init_collection_wrap(cf.gen_unique_str(prefix), shards_num=2)
        for i in range(2):
            df = cf.gen_default_dataframe_data(2 * tmp_nb)
            insert_res, _ = collection_w.insert(df)
            log.debug(collection_w.num_entities)

        collection_w.create_index(ct.default_float_vec_field_name, ct.default_index)
        collection_w.load()
        log.debug(self.utility_wrap.get_query_segment_info(collection_w.name))

        collection_w.compact()
        collection_w.wait_for_compaction_completed()
        c_plans = collection_w.get_compaction_plans()[0]
        assert len(c_plans.plans) == 2
        targets = []
        for plan in c_plans.plans:
            assert len(plan.sources) == 2
            targets.append(plan.target)

        collection_w.release()
        collection_w.load()
        seg_info, _ = self.utility_wrap.get_query_segment_info(collection_w.name)
        loaded_segment_ids = {seg.segmentID for seg in seg_info}
        assert loaded_segment_ids == set(targets), (
            f"loaded segments do not match compaction targets: loaded={loaded_segment_ids}, targets={set(targets)}"
        )

    @pytest.mark.tags(CaseLabel.L1)
    def test_compact_after_index(self):
        """
        target: test compact after create index
        method: 1.insert data into two segments
                2.create index
                3.compact
                4.search
        expected: Verify segment info and index info
        """
        collection_w = self.collection_insert_multi_segments_one_shard(
            prefix, nb_of_segment=ct.default_nb, is_dup=False
        )

        # create index
        collection_w.create_index(ct.default_float_vec_field_name, ct.default_index)
        log.debug(collection_w.index())

        # compact
        collection_w.compact()
        collection_w.wait_for_compaction_completed()
        collection_w.get_compaction_plans(check_task=CheckTasks.check_merge_compact)

        # search
        collection_w.load()
        self.utility_wrap.get_query_segment_info(collection_w.name)
        search_res, _ = collection_w.search(
            cf.gen_vectors(ct.default_nq, ct.default_dim),
            ct.default_float_vec_field_name,
            ct.default_search_params,
            ct.default_limit,
        )
        assert len(search_res) == ct.default_nq
        for hits in search_res:
            assert len(hits) == ct.default_limit

    @pytest.mark.tags(CaseLabel.L1)
    def test_compact_after_binary_index(self):
        """
        target: test compact after create index
        method: 1.insert binary data into two segments
                2.create binary index
                3.compact
                4.search
        expected: Verify segment info and index info
        """
        # create collection with 1 shard and insert 2 segments
        collection_w = self.init_collection_wrap(
            name=cf.gen_unique_str(prefix), shards_num=1, schema=cf.gen_default_binary_collection_schema()
        )
        binary_vectors_by_id = {}
        for i in range(2):
            df, raw_vectors = cf.gen_default_binary_dataframe_data(start=i * ct.default_nb)
            binary_vectors_by_id.update(zip(df[ct.default_int64_field_name].to_list(), raw_vectors, strict=True))
            collection_w.insert(data=df)
            assert collection_w.num_entities == (i + 1) * ct.default_nb

        # create index
        collection_w.create_index(ct.default_binary_vec_field_name, ct.default_binary_index)
        log.debug(collection_w.index())

        # load and search
        collection_w.load()
        search_params = {"metric_type": "JACCARD", "params": {"nprobe": 64}}
        probe_ids = df[ct.default_int64_field_name][: ct.default_nq].to_list()
        probe_vectors = raw_vectors[: ct.default_nq]
        search_res_one, _ = collection_w.search(
            df[ct.default_binary_vec_field_name][: ct.default_nq].to_list(),
            ct.default_binary_vec_field_name,
            search_params,
            ct.default_limit,
        )
        assert len(search_res_one) == ct.default_nq
        assert all(len(hits) == ct.default_limit for hits in search_res_one)

        # compact
        collection_w.compact()
        collection_w.wait_for_compaction_completed()
        c_plans = collection_w.get_compaction_plans(check_task=CheckTasks.check_merge_compact)[0]

        old_segmentIDs = c_plans.plans[0].sources
        # waiting for handoff completed and search
        cost = 180
        start = time()
        while True:
            sleep(1)
            segment_info = self.utility_wrap.get_query_segment_info(collection_w.name)[0]
            if len(segment_info) != 0 and segment_info[0].segmentID not in old_segmentIDs and segment_info[0].is_sorted:
                log.debug(segment_info)
                break
            if time() - start > cost:
                raise MilvusException(1, f"Handoff after compact and index cost more than {cost}s")

        # verify search result
        search_res_two, _ = collection_w.search(
            df[ct.default_binary_vec_field_name][: ct.default_nq].to_list(),
            ct.default_binary_vec_field_name,
            search_params,
            ct.default_limit,
        )
        assert len(search_res_two) == ct.default_nq
        for probe_id, probe_vector, before_hits, after_hits in zip(
            probe_ids, probe_vectors, search_res_one, search_res_two, strict=True
        ):
            assert len(after_hits) == ct.default_limit
            assert len(set(before_hits.ids)) == ct.default_limit
            assert len(set(after_hits.ids)) == ct.default_limit
            assert before_hits.ids[0] == probe_id
            assert after_hits.ids[0] == probe_id
            expected_distances = sorted(
                cf.jaccard(probe_vector, candidate_vector) for candidate_vector in binary_vectors_by_id.values()
            )[: ct.default_limit]
            assert before_hits.distances == pytest.approx(expected_distances)
            assert after_hits.distances == pytest.approx(expected_distances)
            for hits in (before_hits, after_hits):
                for hit_id, hit_distance in zip(hits.ids, hits.distances, strict=True):
                    assert hit_distance == pytest.approx(cf.jaccard(probe_vector, binary_vectors_by_id[hit_id]))

        query_res, _ = collection_w.query(
            f"{ct.default_int64_field_name} in {probe_ids}", output_fields=[ct.default_int64_field_name]
        )
        query_ids = [row[ct.default_int64_field_name] for row in query_res]
        assert len(query_ids) == len(probe_ids)
        assert len(set(query_ids)) == len(probe_ids)
        assert set(query_ids) == set(probe_ids)
        assert collection_w.num_entities == 2 * ct.default_nb

    @pytest.mark.tags(CaseLabel.L1)
    def test_compact_and_index(self):
        """
        target: test compact and create index
        method: 1.insert data into two segments
                2.compact
                3.create index
                4.load and search
        expected: Verify search result and index info
        """
        pytest.skip("Compaction requires segment indexed")
        collection_w = self.collection_insert_multi_segments_one_shard(
            prefix, nb_of_segment=ct.default_nb, is_dup=False
        )

        # compact
        collection_w.compact()
        collection_w.wait_for_compaction_completed()
        collection_w.get_compaction_plans(check_task=CheckTasks.check_merge_compact)

        # create index
        collection_w.create_index(ct.default_float_vec_field_name, ct.default_index)
        log.debug(collection_w.index())

        # search
        collection_w.load()
        search_res, _ = collection_w.search(
            cf.gen_vectors(ct.default_nq, ct.default_dim),
            ct.default_float_vec_field_name,
            ct.default_search_params,
            ct.default_limit,
        )
        assert len(search_res) == ct.default_nq
        for hits in search_res:
            assert len(hits) == ct.default_limit

    @pytest.mark.tags(CaseLabel.L3)
    def test_compact_delete_and_search(self):
        """
        target: test delete and compact segment, and search
        method: 1.create collection and insert
                2.delete part entities
                3.compact
                4.load and search
        expected: Verify search result
        """
        collection_w = self.init_collection_wrap(cf.gen_unique_str(prefix), shards_num=1)
        df = cf.gen_default_dataframe_data()
        insert_res, _ = collection_w.insert(df)

        expr = f"{ct.default_int64_field_name} in {insert_res.primary_keys[: ct.default_nb // 2]}"
        collection_w.delete(expr)
        assert collection_w.num_entities == ct.default_nb

        sleep(ct.compact_retention_duration + 1)
        collection_w.compact()
        collection_w.wait_for_compaction_completed()
        collection_w.get_compaction_plans(check_task=CheckTasks.check_delete_compact)

        # search
        collection_w.load()
        search_res, _ = collection_w.search(
            cf.gen_vectors(ct.default_nq, ct.default_dim),
            ct.default_float_vec_field_name,
            ct.default_search_params,
            ct.default_limit,
            check_task=CheckTasks.check_search_results,
            check_items={
                "nq": ct.default_nq,
                "ids": insert_res.primary_keys[ct.default_nb // 2 :],
                "limit": ct.default_limit,
            },
        )
        collection_w.query("int64 in [0]", check_task=CheckTasks.check_query_empty)

    @pytest.mark.tags(CaseLabel.L0)
    def test_compact_merge_and_search(self):
        """
        target: test compact and search
        method: 1.insert data into two segments
                2.compact
                3.load and search
        expected: Verify search result
        """
        pytest.skip("Compaction requires segment indexed")
        collection_w = self.collection_insert_multi_segments_one_shard(
            prefix, nb_of_segment=ct.default_nb, is_dup=False
        )

        # compact
        collection_w.compact()
        collection_w.wait_for_compaction_completed()
        collection_w.get_compaction_plans(check_task=CheckTasks.check_merge_compact)

        # search
        collection_w.load()
        search_res, _ = collection_w.search(
            cf.gen_vectors(ct.default_nq, ct.default_dim),
            ct.default_float_vec_field_name,
            ct.default_search_params,
            ct.default_limit,
        )
        assert len(search_res) == ct.default_nq
        for hits in search_res:
            assert len(hits) == ct.default_limit

    @pytest.mark.tags(CaseLabel.L2)
    def test_compact_search_after_delete_channel(self):
        """
        target: test search after compact, and queryNode get delete request from channel,
                rather than compacted delta log
        method: 1.insert, flush and load
                2.delete half
                3.compact
                4.search
        expected: No compact, compact get delta log from storage
        """
        collection_w = self.init_collection_wrap(cf.gen_unique_str(prefix), shards_num=1)

        df = cf.gen_default_dataframe_data()
        insert_res, _ = collection_w.insert(df)
        assert collection_w.num_entities == ct.default_nb
        collection_w.create_index(ct.default_float_vec_field_name, ct.default_index)
        collection_w.load()

        expr = f"{ct.default_int64_field_name} in {insert_res.primary_keys[: ct.default_nb // 2]}"
        collection_w.delete(expr)

        collection_w.compact()
        collection_w.wait_for_compaction_completed()
        c_plans = collection_w.get_compaction_plans()[0]
        assert len(c_plans.plans) == 1

        # search
        collection_w.load()
        search_res, _ = collection_w.search(
            cf.gen_vectors(ct.default_nq, ct.default_dim),
            ct.default_float_vec_field_name,
            ct.default_search_params,
            ct.default_limit,
            check_task=CheckTasks.check_search_results,
            check_items={
                "nq": ct.default_nq,
                "ids": insert_res.primary_keys[ct.default_nb // 2 :],
                "limit": ct.default_limit,
            },
        )

    @pytest.mark.tags(CaseLabel.L3)
    def test_compact_repeatedly_delete_same_id(self):
        """
        target: test compact after repeatedly delete same entity
        method: 1.Create and insert entities
                2.repeatedly delete the same id
                3.compact
        expected: No exception or delta log just delete one
        """
        collection_w = self.init_collection_wrap(cf.gen_unique_str(prefix), shards_num=1)

        df = cf.gen_default_dataframe_data()
        insert_res, _ = collection_w.insert(df)
        expr = f"{ct.default_int64_field_name} in [0]"

        for _ in range(100):
            collection_w.delete(expr=expr)
        assert collection_w.num_entities == ct.default_nb

        sleep(ct.compact_retention_duration + 1)
        collection_w.compact()
        collection_w.wait_for_compaction_completed()
        collection_w.get_compaction_plans(check_task=CheckTasks.check_delete_compact)

        collection_w.load()
        collection_w.query(expr, check_task=CheckTasks.check_query_empty)

    @pytest.mark.tags(CaseLabel.L0)
    def test_compact_merge_two_segments(self):
        """
        target: test compact merge two segments
        method: 1.create with shard_num=1
                2.insert and flush
                3.insert and flush again
                4.compact
                5.load
        expected: Verify segments are merged
        """
        num_of_segment = 2
        # create collection shard_num=1, insert 2 segments, each with tmp_nb entities
        collection_w = self.collection_insert_multi_segments_one_shard(prefix, num_of_segment, tmp_nb)
        collection_w.create_index(ct.default_float_vec_field_name, ct.default_index)
        log.debug(collection_w.index())

        collection_w.compact()
        collection_w.wait_for_compaction_completed()
        collection_w.get_compaction_plans(check_task=CheckTasks.check_merge_compact)

        # the two segments being merged into one is already asserted by check_merge_compact above

        # verify queryNode load the compacted segments
        collection_w.load()

        start = time()
        cost = 180
        while True:
            sleep(1)
            segments_info = self.utility_wrap.get_query_segment_info(collection_w.name)[0]

            # verify segments reaches threshold, auto-merge ten segments into one
            if len(segments_info) == 1:
                break
            end = time()
            if end - start > cost:
                raise MilvusException(1, "Compact merge two segments more than 180s")
        # target --sort--> querySegment
        # assert c_plans.plans[0].target == segments_info[0].segmentID

    @pytest.mark.tags(CaseLabel.L2)
    def test_compact_no_merge(self):
        """
        target: test compact when no segments merge
        method: 1.create with shard_num=1
                2.insert and flush
                3.compact and search
        expected: No exception and compact plans
        """
        # create collection
        collection_w = self.init_collection_wrap(name=cf.gen_unique_str(prefix), shards_num=1)
        df = cf.gen_default_dataframe_data(tmp_nb)
        collection_w.insert(df)
        assert collection_w.num_entities == tmp_nb

        collection_w.create_index(ct.default_float_vec_field_name, ct.default_index)
        collection_w.load()

        seg_before, _ = self.utility_wrap.get_query_segment_info(collection_w.name)

        collection_w.compact()
        collection_w.wait_for_compaction_completed()
        c_plans, _ = collection_w.get_compaction_plans()
        assert len(c_plans.plans) == 1
        assert [seg_before[0].segmentID] == c_plans.plans[0].sources

        collection_w.release()
        collection_w.load()
        seg_after, _ = self.utility_wrap.get_query_segment_info(collection_w.name)
        # target --sort--> querySegment
        # assert seg_after[0].segmentID == c_plans.plans[0].target

    @pytest.mark.tags(CaseLabel.L1)
    def test_compact_manual_and_auto(self):
        """
        target: test compact manual and auto
        method: 1.create with shard_num=1
                2.insert one and flush (11 times)
                3.compact
                4.load and search
        expected: Verify segments info
        """
        # greater than auto-merge threshold 10
        pytest.skip(
            "DataCoord: for A, B -> C, will not compact segment C before A, B GCed, no method to check whether a segment is GCed"
        )
        num_of_segment = ct.compact_segment_num_threshold + 1

        # create collection shard_num=1, insert 11 segments, each with one entity
        collection_w = self.collection_insert_multi_segments_one_shard(prefix, num_of_segment=num_of_segment)
        collection_w.create_index(ct.default_float_vec_field_name, ct.default_index)
        log.debug(collection_w.index())

        # waiting for auto compaction finished
        sleep(60)

        collection_w.compact()
        collection_w.wait_for_compaction_completed()
        c_plans, _ = collection_w.get_compaction_plans(
            check_task=CheckTasks.check_merge_compact, check_items={"segment_num": 2}
        )

        collection_w.load()
        start = time()
        cost = 180
        while True:
            sleep(1)
            segments_info = self.utility_wrap.get_query_segment_info(collection_w.name)[0]

            # verify segments reaches threshold, auto-merge ten segments into one
            if len(segments_info) == 1:
                break
            end = time()
            if end - start > cost:
                raise MilvusException(1, "Compact auto and manual more than 180s")
        assert segments_info[0].segmentID == c_plans.plans[0].target

    @pytest.mark.tags(CaseLabel.L1)
    def test_compact_merge_multi_segments(self):
        """
        target: test compact and merge multi small segments
        method: 1.create with shard_num=1
                2.insert one and flush (less than threshold)
                3.compact
                4.load and search
        expected: Verify segments info
        """
        # less than auto-merge threshold 10
        num_of_segment = ct.compact_segment_num_threshold - 1

        # create collection shard_num=1, insert 11 segments, each with one entity
        collection_w = self.collection_insert_multi_segments_one_shard(prefix, num_of_segment=num_of_segment)
        collection_w.create_index(ct.default_float_vec_field_name, ct.default_index)
        log.debug(collection_w.index())

        collection_w.compact()
        collection_w.wait_for_compaction_completed()

        c_plans = collection_w.get_compaction_plans(
            check_task=CheckTasks.check_merge_compact, check_items={"segment_num": num_of_segment}
        )[0]
        target = c_plans.plans[0].target

        collection_w.load()
        cost = 180
        start = time()
        while True:
            sleep(1)
            segments_info = self.utility_wrap.get_query_segment_info(collection_w.name)[0]

            # verify segments reaches threshold, auto-merge ten segments into one
            if len(segments_info) == 1:
                break
            end = time()
            if end - start > cost:
                raise MilvusException(1, "Compact merge multiple segments more than 180s")
        replicas = collection_w.get_replicas()[0]
        replica_num = len(replicas.groups)
        assert len(segments_info) == 1 * replica_num
        # target --sort--> querySegment
        # assert segments_info[0].segmentID == target

    @pytest.mark.tags(CaseLabel.L2)
    def test_compact_threshold_auto_merge(self):
        """
        target: test num (segment_size < 1/2Max) reaches auto-merge threshold 10
        method: 1.create with shard_num=1
                2.insert flush 10 times (merge threshold 10)
                3.wait for compaction, load
        expected: Get query segments info to verify segments auto-merged into one
        """
        threshold = ct.compact_segment_num_threshold

        # create collection shard_num=1, insert 10 segments, each with one entity
        collection_w = self.collection_insert_multi_segments_one_shard(prefix, num_of_segment=threshold)
        collection_w.create_index(ct.default_float_vec_field_name, ct.default_index)
        log.debug(collection_w.index())

        # Estimated auto-merging takes 30s
        cost = 180
        collection_w.load()
        replicas = collection_w.get_replicas()[0]
        replica_num = len(replicas.groups)
        start = time()
        while True:
            sleep(1)
            segments_info = self.utility_wrap.get_query_segment_info(collection_w.name)[0]

            # verify segments reaches threshold, auto-merge ten segments into one
            if len(segments_info) == 1 * replica_num:
                break
            end = time()
            if end - start > cost:
                raise MilvusException(1, "Compact auto-merge more than 180s")

    @pytest.mark.tags(CaseLabel.L2)
    def test_compact_less_threshold_no_merge(self):
        """
        target: test compact the num of segments that size less than 1/2Max, does not reach the threshold
        method: 1.create collection with shard_num = 1
                2.insert flush 9 times (segments threshold 10)
                3.after a while, load
        expected: Verify segments are not merged
        """
        less_threshold = ct.compact_segment_num_threshold - 1

        # create collection shard_num=1, insert 9 segments, each with one entity
        collection_w = self.collection_insert_multi_segments_one_shard(prefix, num_of_segment=less_threshold)

        # create index
        collection_w.create_index(ct.default_float_vec_field_name, ct.default_index)

        # load and verify no auto-merge
        collection_w.load()
        replicas = collection_w.get_replicas()[0]
        replica_num = len(replicas.groups)
        segments_info = self.utility_wrap.get_query_segment_info(collection_w.name)[0]
        assert len(segments_info) == less_threshold * replica_num

    @pytest.mark.skip(reason="Todo")
    @pytest.mark.tags(CaseLabel.L2)
    def test_compact_multi_collections(self):
        """
        target: test compact multi collections with merge
        method: create 50 collections, add entities into them and compact in turn
        expected: No exception
        """
        pass

    @pytest.mark.tags(CaseLabel.L1)
    def test_compact_and_insert(self):
        """
        target: test insert after compact
        method: 1.create and insert with flush
                2.delete and compact
                3.insert new data
                4.load and search
        expected: Verify search result and segment info
        """
        # create collection shard_num=1, insert 2 segments, each with tmp_nb entities
        collection_w = self.collection_insert_multi_segments_one_shard(prefix, nb_of_segment=tmp_nb)

        # compact two segments
        collection_w.compact()
        collection_w.wait_for_compaction_completed()
        collection_w.get_compaction_plans()

        # insert new data, verify insert flush successfully
        df = cf.gen_default_dataframe_data(tmp_nb)
        collection_w.insert(df)
        assert collection_w.num_entities == tmp_nb * 3

    @pytest.mark.tags(CaseLabel.L1)
    def test_compact_and_delete(self):
        """
        target: test delete after compact
        method: 1.delete half and compact
                2.load and query
                3.delete and query
        expected: Verify deleted ids
        """
        # init collection with one shard, insert into two segments
        collection_w = self.collection_insert_multi_segments_one_shard(prefix, is_dup=False)
        collection_w.create_index(ct.default_float_vec_field_name, ct.default_index)
        log.debug(collection_w.index())
        # compact and complete
        collection_w.compact()
        collection_w.wait_for_compaction_completed()
        collection_w.get_compaction_plans(check_task=CheckTasks.check_merge_compact)

        # delete and query
        expr = f"{ct.default_int64_field_name} in {[0]}"
        collection_w.delete(expr)
        collection_w.load()
        collection_w.query(expr, check_task=CheckTasks.check_query_empty)

        expr_1 = f"{ct.default_int64_field_name} in {[1]}"
        collection_w.query(
            expr_1,
            check_task=CheckTasks.check_query_results,
            check_items={
                "exp_res": [{"int64": 1}],
                "pk_name": collection_w.primary_field.name,
            },
        )

    @pytest.mark.tags(CaseLabel.L1)
    def test_compact_cross_shards(self):
        """
        target: test compact cross shards
        method: 1.create with shard_num=2
                2.insert once and flush (two segments, belonging to two shards)
                3.compact and completed
        expected: Verify compact plan sources only one segment
        """
        # insert into two segments with two shard
        collection_w = self.init_collection_wrap(name=cf.gen_unique_str(prefix), shards_num=2)
        df = cf.gen_default_dataframe_data(tmp_nb)
        collection_w.insert(df)
        assert collection_w.num_entities == tmp_nb
        collection_w.create_index(ct.default_float_vec_field_name, ct.default_index)
        log.debug(collection_w.index())

        # compact
        collection_w.compact()
        collection_w.wait_for_compaction_completed()
        c_plans = collection_w.get_compaction_plans()[0]

        # Actually no merged
        assert len(c_plans.plans) == 2
        for plan in c_plans.plans:
            assert len(plan.sources) == 1

    @pytest.mark.tags(CaseLabel.L3)
    def test_compact_delete_cross_shards(self):
        """
        target: test delete compact cross different shards
        method: 1.create with 2 shards
                2.insert entities into 2 segments
                3.delete one entity from each segment
                4.call compact and get compact plan
        expected: Generate compaction plan for each segment
        """
        shards_num = 2
        # insert into two segments with two shard
        collection_w = self.init_collection_wrap(name=cf.gen_unique_str(prefix), shards_num=shards_num)
        df = cf.gen_default_dataframe_data(tmp_nb)
        collection_w.insert(df)
        expr = f"{ct.default_int64_field_name} in [0, 99]"
        collection_w.delete(expr)
        assert collection_w.num_entities == tmp_nb
        collection_w.create_index(ct.default_float_vec_field_name, ct.default_index)
        log.debug(collection_w.index())

        # compact
        sleep(ct.compact_retention_duration + 1)
        collection_w.compact()
        collection_w.wait_for_compaction_completed()
        collection_w.get_compaction_plans(
            check_task=CheckTasks.check_delete_compact, check_items={"plans_num": shards_num}
        )

    @pytest.mark.tags(CaseLabel.L1)
    def test_compact_cross_partition(self):
        """
        target: test compact cross partitions
        method: 1.create with shard_num=1
                2.create partition and insert, flush
                3.insert _default partition and flush
                4.compact
        expected: Verify two independent compaction plans
        """
        # create collection and partition
        collection_w = self.init_collection_wrap(name=cf.gen_unique_str(prefix), shards_num=1)
        partition_w = self.init_partition_wrap(collection_wrap=collection_w)

        # insert
        df = cf.gen_default_dataframe_data(tmp_nb)
        collection_w.insert(df)
        assert collection_w.num_entities == tmp_nb
        partition_w.insert(df)
        assert collection_w.num_entities == tmp_nb * 2

        # compaction only applied to indexed segments.
        collection_w.create_index(ct.default_float_vec_field_name, ct.default_index)
        log.debug(collection_w.index())

        # compact
        collection_w.compact()
        collection_w.wait_for_compaction_completed()
        c_plans = collection_w.get_compaction_plans()[0]

        # since manual compaction, segment should be compacted any way
        assert len(c_plans.plans) == 2
        for plan in c_plans.plans:
            assert len(plan.sources) == 1

    @pytest.mark.tags(CaseLabel.L1)
    def test_compact_during_insert(self):
        """Sustain inserts across compaction, then verify every original and new PK."""
        num_of_segment = 9
        initial_rows = ct.default_nb * num_of_segment
        collection_w = self.collection_insert_multi_segments_one_shard(
            prefix,
            num_of_segment=num_of_segment,
            nb_of_segment=ct.default_nb,
            is_dup=False,
            collection_properties={"collection.autocompaction.enabled": "false"},
        )
        collection_w.create_index(ct.default_float_vec_field_name, ct.default_index)
        template = cf.gen_default_dataframe_data(nb=tmp_nb)
        inserted_rows = 0

        def insert_batch(timeout):
            nonlocal inserted_rows
            df = template.copy()
            start = initial_rows + inserted_rows
            df[ct.default_int64_field_name] = range(start, start + tmp_nb)
            result = collection_w.insert(df, timeout=timeout)[0]
            assert result.insert_count == tmp_nb
            assert list(result.primary_keys) == list(range(start, start + tmp_nb))
            inserted_rows += tmp_nb

        observed = run_compaction_with_workload(
            lambda timeout: collection_w.compact(timeout=timeout),
            lambda timeout: collection_w.get_detailed_compaction_state(timeout=timeout),
            insert_batch,
        )
        assert inserted_rows > 0
        collection_w.get_compaction_plans()
        collection_w.flush()
        assert self.utility_wrap.wait_for_index_building_complete(collection_w.name, timeout=240)[0]
        collection_w.compact()
        collection_w.wait_for_compaction_completed()
        collection_w.get_compaction_plans()
        collection_w.load()

        search_res = collection_w.search(
            [template[ct.default_float_vec_field_name][0]],
            ct.default_float_vec_field_name,
            ct.default_search_params,
            ct.default_limit,
        )[0]
        assert len(search_res[0]) == ct.default_limit
        # Bounded PK ranges avoid the query result limit as the sustained
        # workload grows. Check originals too, including duplicate/missing IDs.
        total_rows = initial_rows + inserted_rows
        for start in range(0, total_rows, 1000):
            end = min(start + 1000, total_rows)
            rows = collection_w.query(
                f"{ct.default_int64_field_name} >= {start} && {ct.default_int64_field_name} < {end}",
                output_fields=[ct.default_int64_field_name],
                consistency_level="Strong",
            )[0]
            ids = [row[ct.default_int64_field_name] for row in rows]
            assert len(ids) == end - start
            assert set(ids) == set(range(start, end))
        require_observed_overlap(observed)

    @pytest.mark.tags(CaseLabel.L2)
    def test_compact_during_index(self):
        """Sustain CreateIndex requests across compaction and check indexed outputs."""
        num_of_segment = 9
        collection_w = self.collection_insert_multi_segments_one_shard(
            prefix,
            num_of_segment=num_of_segment,
            nb_of_segment=ct.default_nb,
            is_dup=False,
            collection_properties={"collection.autocompaction.enabled": "false"},
        )
        deadline = monotonic() + 300
        # Flush completion alone does not mean sorting has finished. Capture
        # stable inputs before the index workload and the manual Mix trigger.
        sources = collection_w.wait_for_compaction_eligible_segments(
            minimum_segment_count=num_of_segment,
            timeout=min(60, remaining_time(deadline, "waiting for stable sorted inputs")),
        )
        source_ids = {segment.segment_id for segment in sources}
        assert len(source_ids) == len(sources), f"Duplicate source IDs: {sources}"
        assert sum(segment.num_rows for segment in sources) == num_of_segment * ct.default_nb

        def create_index(timeout):
            # sync=False acknowledges the request; actual build completion is
            # checked below. Repeated requests are idempotent, not rebuilds.
            collection_w.create_index(ct.default_float_vec_field_name, ct.default_index, timeout=timeout, sync=False)

        observed = run_compaction_with_workload(
            lambda timeout: collection_w.compact(timeout=timeout),
            lambda timeout: collection_w.get_detailed_compaction_state(timeout=timeout),
            create_index,
            timeout=remaining_time(deadline, "running index workload"),
        )
        assert collection_w.index(timeout=remaining_time(deadline, "checking index"))[0].params == ct.default_index
        compaction_plans = collection_w.get_compaction_plans(
            timeout=remaining_time(deadline, "getting compaction plans")
        )[0]

        assert self.utility_wrap.wait_for_index_building_complete(
            collection_w.name,
            timeout=remaining_time(deadline, "waiting for index build completion"),
        )[0]
        raw_target_ids = [plan.target for plan in compaction_plans.plans]
        target_ids = set(raw_target_ids)
        assert raw_target_ids, f"Compaction completed without plans: {compaction_plans}"
        assert len(raw_target_ids) == len(target_ids), f"Compaction targets must be unique: {compaction_plans}"

        selected_sources = [source for plan in compaction_plans.plans for source in plan.sources]
        selected_source_ids = set(selected_sources)
        assert selected_sources, f"Compaction has no selected sources: {compaction_plans}"
        assert len(selected_sources) == len(selected_source_ids), f"Repeated plan sources: {compaction_plans}"
        assert selected_source_ids <= source_ids, f"Unknown plan sources: {compaction_plans}, before={source_ids}"
        assert target_ids.isdisjoint(source_ids), f"Targets reuse source IDs: {compaction_plans}"
        # Manual Mix may select only a subset, even with stable sorted inputs.
        expected_loaded_ids = (source_ids - selected_source_ids) | target_ids

        collection_w.load(timeout=remaining_time(deadline, "loading indexed compaction targets"))
        seg_info = []
        while True:
            seg_info = self.utility_wrap.get_query_segment_info(
                collection_w.name,
                timeout=remaining_time(deadline, "waiting for compaction target handoff"),
            )[0]
            loaded_ids = [segment.segmentID for segment in seg_info]
            targets_loaded_and_indexed = (
                len(loaded_ids) == len(set(loaded_ids))
                and set(loaded_ids) == expected_loaded_ids
                and set(loaded_ids).isdisjoint(selected_source_ids)
                and all(segment.indexID > 0 and segment.index_name and segment.nodeIds for segment in seg_info)
            )
            if targets_loaded_and_indexed:
                break
            sleep(min(1, remaining_time(deadline, "waiting for indexed compaction targets")))

        search_res = collection_w.search(
            cf.gen_vectors(1, ct.default_dim),
            ct.default_float_vec_field_name,
            ct.default_search_params,
            ct.default_limit,
        )[0]
        assert len(search_res[0]) == ct.default_limit
        total_rows = num_of_segment * ct.default_nb
        for start in range(0, total_rows, 1000):
            end = min(start + 1000, total_rows)
            rows = collection_w.query(
                f"{ct.default_int64_field_name} >= {start} && {ct.default_int64_field_name} < {end}",
                output_fields=[ct.default_int64_field_name],
                consistency_level="Strong",
                timeout=remaining_time(deadline, "checking indexed collection PKs"),
            )[0]
            ids = [row[ct.default_int64_field_name] for row in rows]
            assert len(ids) == end - start
            assert set(ids) == set(range(start, end))
        require_observed_overlap(observed)

    @pytest.mark.tags(CaseLabel.L2)
    def test_compact_during_search(self):
        """Sustain checked searches across compaction and validate the final result."""
        num_of_segment = ct.compact_segment_num_threshold - 1
        collection_w = self.collection_insert_multi_segments_one_shard(
            prefix,
            num_of_segment=num_of_segment,
            nb_of_segment=100,
            is_dup=False,
            collection_properties={"collection.autocompaction.enabled": "false"},
        )
        collection_w.create_index(ct.default_float_vec_field_name, ct.default_index)
        collection_w.load()
        vectors = cf.gen_vectors(1, ct.default_dim)
        expected_ids = set(range(num_of_segment * 100))

        def search(timeout):
            result = collection_w.search(
                vectors,
                ct.default_float_vec_field_name,
                ct.default_search_params,
                ct.default_limit,
                timeout=timeout,
            )[0]
            ids = list(result[0].ids)
            assert len(ids) == len(set(ids)) == ct.default_limit
            assert set(ids) <= expected_ids

        observed = run_compaction_with_workload(
            lambda timeout: collection_w.compact(timeout=timeout),
            lambda timeout: collection_w.get_detailed_compaction_state(timeout=timeout),
            search,
        )
        collection_w.get_compaction_plans(
            check_task=CheckTasks.check_merge_compact,
            check_items={"segment_num": num_of_segment},
        )
        search(60)
        require_observed_overlap(observed)