docstr: target: test compact during insert and flush
  method: 1.insert entities into multi segments
          2.start a thread to load and search
          3.compact collection
  expected: Search and compact both successfully
docstr_tok: ["target",":","test","compact","during","insert","and","flush","method",":","1.insert","entities","into","multi","segments","2.start","a","thread","to","load","and","search","3.compact","collection","expected",":","Search","and","compact","both","successfully"]
code: def test_compact_during_insert(self):
      """
      target: test compact during insert and flush
      method: 1.insert entities into multi segments
              2.start a thread to load and search
              3.compact collection
      expected: Search and compact both successfully
      """
      collection_w = self.collection_insert_multi_segments_one_shard(
          prefix, nb_of_segment=ct.default_nb, is_dup=False
      )
      collection_w.create_index(
          ct.default_float_vec_field_name, ct.default_index
      )
      log.debug(collection_w.index())
      df = cf.gen_default_dataframe_data(start=ct.default_nb * 2)
  
      def do_flush():
          collection_w.insert(df)
          log.debug(collection_w.num_entities)
  
      # compact during insert
      t = threading.Thread(target=do_flush, args=())
      t.start()
      collection_w.compact()
      collection_w.wait_for_compaction_completed()
      collection_w.get_compaction_plans()
      t.join()
  
      # waitting for new segment index and compact
      index_cost = 240
      start = time()
      while True:
          sleep(10)
          collection_w.load()
          # new segment compacted
          seg_info = self.utility_wrap.get_query_segment_info(
              collection_w.name
          )[0]
          if len(seg_info) == 2:
              break
          end = time()
          collection_w.release()
          if end - start > index_cost:
              raise MilvusException(
                  1,
                  f"Waiting more than {index_cost}s for the new segment indexed",
              )
  
      # compact new segment
      collection_w.compact()
      collection_w.wait_for_compaction_completed()
      collection_w.get_compaction_plans()
  
      # waitting for new segment index and compact
      compact_cost = 180
      start = time()
      while True:
          sleep(1)
          collection_w.load()
          # new segment compacted
          seg_info = self.utility_wrap.get_query_segment_info(
              collection_w.name
          )[0]
          if len(seg_info) == 1:
              break
          end = time()
          collection_w.release()
          if end - start > compact_cost:
              raise MilvusException(
                  1,
                  f"Waiting more than {compact_cost}s for the new target segment to load",
              )
  
      # search
      search_res, _ = collection_w.search(
          [df[ct.default_float_vec_field_name][0]],
          ct.default_float_vec_field_name,
          ct.default_search_params,
          ct.default_limit,
      )
      assert len(search_res[0]) == ct.default_limit
  
imports: import threading
  from time import time, sleep
  import pytest
  from pymilvus.grpc_gen.common_pb2 import SegmentState
  from pymilvus.exceptions import MilvusException
  from base.client_base import TestcaseBase
  from common import common_func
  from common import common_type
  from common.common_type import CaseLabel, CheckTasks
  from utils.util_log import test_log
label_desc: Use image node_id 23 for calling the TestCompactionOperation obj's underlying member method code with example usage: obj.test_compact_during_insert() without return types