docstr: collection schema: [str_pk, float_vector,
                  float_scalar, int_scalar, string_scalar, bool_scalar]
  Steps:
  1. create collection with string primary key
  2. create index and load collection
  3. import data
  4. verify the data entities
  5. verify index status
  6. verify search and query
code: def test_string_pk_float_vector_multi_scalars(
      self, is_row_based, fields_num_in_file, dim, entities
  ):
      """
      collection schema: [str_pk, float_vector,
                      float_scalar, int_scalar, string_scalar, bool_scalar]
      Steps:
      1. create collection with string primary key
      2. create index and load collection
      3. import data
      4. verify the data entities
      5. verify index status
      6. verify search and query
      """
      string_pk = True
      auto_id = False
      files = prepare_bulk_insert_json_files(
          minio_endpoint=self.minio_endpoint,
          bucket_name=self.bucket_name,
          is_row_based=is_row_based,
          rows=entities,
          dim=dim,
          auto_id=auto_id,
          str_pk=string_pk,
          data_fields=default_multi_fields,
      )
      additional_field = "int_scalar_add"
      self._connect()
      c_name = cf.gen_unique_str("bulk_insert")
      fields = [
          cf.gen_string_field(name=df.pk_field, is_primary=True),
          cf.gen_float_vec_field(name=df.vec_field, dim=dim),
          cf.gen_int32_field(name=df.int_field),
          cf.gen_string_field(name=df.string_field),
          cf.gen_bool_field(name=df.bool_field),
          cf.gen_float_field(name=df.float_field),
      ]
      if fields_num_in_file == "more":
          fields.pop()
      elif fields_num_in_file == "less":
          fields.append(cf.gen_int32_field(name=additional_field))
      schema = cf.gen_collection_schema(fields=fields, auto_id=auto_id)
      self.collection_wrap.init_collection(c_name, schema=schema)
      # build index
      index_params = ct.default_index
      self.collection_wrap.create_index(
          field_name=df.vec_field, index_params=index_params
      )
      # load collection
      self.collection_wrap.load()
      # import data
      t0 = time.time()
      task_id, _ = self.utility_wrap.do_bulk_insert(
          collection_name=c_name, files=files
      )
      logging.info(f"bulk insert task ids:{task_id}")
      (
          success,
          states,
      ) = self.utility_wrap.wait_for_bulk_insert_tasks_completed(
          task_ids=[task_id], timeout=90
      )
      tt = time.time() - t0
      log.info(f"bulk insert state:{success} in {tt}")
      if fields_num_in_file in ["less", "more"]:
          assert not success  # TODO: check error msg
          if is_row_based:
              if fields_num_in_file == "less":
                  failed_reason = (
                      f"value of field '{additional_field}' is missed"
                  )
              else:
                  failed_reason = f"field '{df.float_field}' is not defined in collection schema"
          else:
              failed_reason = "is not equal to other fields"
          for state in states.values():
              assert state.state_name in [
                  "Failed",
                  "Failed and cleaned",
              ]
              assert failed_reason in state.infos.get(
                  "failed_reason", ""
              )
      else:
          assert success
          log.info(
              f" collection entities: {self.collection_wrap.num_entities}"
          )
          assert self.collection_wrap.num_entities == entities
          # verify index
          success = self.utility_wrap.wait_index_build_completed(c_name)
          assert success
          # verify search and query
          log.info(f"wait for load finished and be ready for search")
          self.collection_wrap.load(_refresh=True)
          time.sleep(10)
          search_data = cf.gen_vectors(1, dim)
          search_params = ct.default_search_params
          res, _ = self.collection_wrap.search(
              search_data,
              df.vec_field,
              param=search_params,
              limit=1,
              check_task=CheckTasks.check_search_results,
              check_items={"nq": 1, "limit": 1},
          )
          for hits in res:
              ids = hits.ids
              expr = f"{df.pk_field} in {ids}"
              expr = expr.replace("'", '"')
              results, _ = self.collection_wrap.query(expr=expr)
              assert len(results) == len(ids)
  
imports: import logging
  import time
  import pytest
  import random
  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_k8s import get_pod_ip_name_pairs, get_milvus_instance_name, get_milvus_deploy_tool
  from utils.util_log import test_log
  from common.bulk_insert_data import prepare_bulk_insert_json_files, prepare_bulk_insert_numpy_files, DataField, DataErrorType
label_desc: Use image node_id 8 for calling the TestBulkInsert obj's underlying member method code with example usage: obj.test_string_pk_float_vector_multi_scalars(is_row_based, fields_num_in_file, dim, entities) without return types