docstr: 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
docstr_tok: ["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"]
code: 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 > 180:
              raise MilvusException(
                  1, "Auto delete ratio compaction cost more than 180s"
              )
          sleep(1)
  
      collection_w.load()
      collection_w.query(
          ratio_expr, check_items=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_items={"exp_res": res},
      )
  
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 8 for calling the TestCompactionParams obj's underlying member method code with example usage: obj.test_compact_delete_ratio() without return types