docstr: collection: auto_id, customized_id
  collection schema: [pk, float_vector, int64, varchar, bool, float]
  Step:
  1. create collection with partition key enabled
  2. import data
  3. verify the data entities equal the import data and distributed by values of partition key field
  4. load the collection
  5. verify search successfully
  6. verify query successfully
docstr_tok: ["collection",":","auto_id",",","customized_id","collection","schema",":","[","pk",",","float_vector",",","int64",",","varchar",",","bool",",","float","]","Step",":","1.","create","collection","with","partition","key","enabled","2.","import","data","3.","verify","the","data","entities","equal","the","import","data","and","distributed","by","values","of","partition","key","field","4.","load","the","collection","5.","verify","search","successfully","6.","verify","query","successfully"]
code: def test_partition_key_on_csv_file(
      self, auto_id, partition_key_field
  ):
      """
      collection: auto_id, customized_id
      collection schema: [pk, float_vector, int64, varchar, bool, float]
      Step:
      1. create collection with partition key enabled
      2. import data
      3. verify the data entities equal the import data and distributed by values of partition key field
      4. load the collection
      5. verify search successfully
      6. verify query successfully
      """
      dim = 12
      entities = 200
      files = prepare_bulk_insert_csv_files(
          minio_endpoint=self.minio_endpoint,
          bucket_name=self.bucket_name,
          rows=entities,
          dim=dim,
          auto_id=auto_id,
          data_fields=default_multi_fields,
          force=True,
      )
      self._connect()
      c_name = cf.gen_unique_str("bulk_parkey")
      fields = [
          cf.gen_int64_field(name=df.pk_field, is_primary=True),
          cf.gen_float_vec_field(name=df.vec_field, dim=dim),
          cf.gen_int64_field(
              name=df.int_field,
              is_partition_key=(partition_key_field == df.int_field),
          ),
          cf.gen_string_field(
              name=df.string_field,
              is_partition_key=(partition_key_field == df.string_field),
          ),
          cf.gen_bool_field(name=df.bool_field),
          cf.gen_float_field(name=df.float_field),
      ]
      schema = cf.gen_collection_schema(fields=fields, auto_id=auto_id)
      self.collection_wrap.init_collection(
          c_name, schema=schema, num_partitions=10
      )
      assert len(self.collection_wrap.partitions) == 10
  
      # import data
      t0 = time.time()
      task_id, _ = self.utility_wrap.do_bulk_insert(
          collection_name=c_name,
          partition_name=None,
          files=files,
      )
      logging.info(f"bulk insert task id:{task_id}")
      (
          success,
          _,
      ) = self.utility_wrap.wait_for_bulk_insert_tasks_completed(
          task_ids=[task_id], timeout=300
      )
      tt = time.time() - t0
      log.info(f"bulk insert state:{success} in {tt}")
      assert success
  
      num_entities = self.collection_wrap.num_entities
      log.info(f" collection entities: {num_entities}")
      assert num_entities == entities
  
      # verify imported data is available for search
      index_params = ct.default_index
      self.collection_wrap.create_index(
          field_name=df.vec_field, index_params=index_params
      )
      self.collection_wrap.load()
      log.info(f"wait for load finished and be ready for search")
      time.sleep(10)
      log.info(
          f"query seg info: {self.utility_wrap.get_query_segment_info(c_name)[0]}"
      )
      nq = 2
      topk = 2
      search_data = cf.gen_vectors(nq, dim)
      search_params = ct.default_search_params
      res, _ = self.collection_wrap.search(
          search_data,
          df.vec_field,
          param=search_params,
          limit=topk,
          check_task=CheckTasks.check_search_results,
          check_items={"nq": nq, "limit": topk},
      )
      for hits in res:
          ids = hits.ids
          results, _ = self.collection_wrap.query(
              expr=f"{df.pk_field} in {ids}"
          )
          assert len(results) == len(ids)
  
      # verify data was bulk inserted into different partitions
      num_entities = 0
      empty_partition_num = 0
      for p in self.collection_wrap.partitions:
          if p.num_entities == 0:
              empty_partition_num += 1
          num_entities += p.num_entities
      assert num_entities == entities
  
imports: import logging
  import time
  import pytest
  from pymilvus import DataType
  import numpy
  from pathlib import Path
  from base.client_base import TestcaseBase
  from common import common_func
  from common import common_type
  from common.milvus_sys import MilvusSys
  from common.common_type import CaseLabel, CheckTasks
  from utils.util_log import test_log
  from common.bulk_insert_data import prepare_bulk_insert_json_files, prepare_bulk_insert_new_json_files, prepare_bulk_insert_numpy_files, prepare_bulk_insert_parquet_files, prepare_bulk_insert_csv_files, DataField
label_desc: Use image node_id 15 for calling the TestBulkInsert obj's underlying member method code with example usage: obj.test_partition_key_on_csv_file(auto_id, partition_key_field) without return types