milvus-io--milvus/tests/python_client/milvus_client/test_milvus_client_insert.py

# ruff: noqa: E712,E731,F401,F403,F405,F541,F841,I001,UP031,UP032,W291,W292,W293
import pytest
import numpy as np

from base.client_v2_base import TestMilvusClientV2Base
from utils.util_log import test_log as log
from common import common_func as cf
from common import common_type as ct
from common.common_type import CaseLabel, CheckTasks
from utils.util_pymilvus import *

prefix = "client_insert"
epsilon = ct.epsilon
default_nb = ct.default_nb
default_nb_medium = ct.default_nb_medium
default_nq = ct.default_nq
default_dim = ct.default_dim
default_limit = ct.default_limit
default_search_exp = "id >= 0"
exp_res = "exp_res"
default_search_string_exp = 'varchar >= "0"'
default_search_mix_exp = 'int64 >= 0 && varchar >= "0"'
default_invaild_string_exp = "varchar >= 0"
default_json_search_exp = 'json_field["number"] >= 0'
perfix_expr = 'varchar like "0%"'
default_search_field = ct.default_float_vec_field_name
default_search_params = ct.default_search_params
default_primary_key_field_name = "id"
default_vector_field_name = "vector"
default_dynamic_field_name = "field_new"
default_float_field_name = ct.default_float_field_name
default_bool_field_name = ct.default_bool_field_name
default_string_field_name = ct.default_string_field_name
default_int32_array_field_name = ct.default_int32_array_field_name
default_string_array_field_name = ct.default_string_array_field_name
default_int32_field_name = ct.default_int32_field_name
default_int32_value = ct.default_int32_value


class TestMilvusClientInsertInvalid(TestMilvusClientV2Base):
    """Test case of search interface"""

    @pytest.fixture(scope="function", params=[False, True])
    def auto_id(self, request):
        yield request.param

    @pytest.fixture(scope="function", params=["COSINE", "L2"])
    def metric_type(self, request):
        yield request.param

    """
    ******************************************************************
    #  The following are invalid base cases
    ******************************************************************
    """

    @pytest.mark.skip(reason="duplicate with test_insert_without_connection")
    @pytest.mark.tags(CaseLabel.L2)
    def test_milvus_client_insert_after_client_closed(self):
        """
        target: test insert after client is closed
        method: insert after client is closed
        expected: raise exception
        """
        client = self._client(alias="my_client")
        collection_name = cf.gen_collection_name_by_testcase_name()
        self.create_collection(client, collection_name, default_dim)
        self.close(client)

        data = cf.gen_default_list_data(10)
        error = {ct.err_code: 999, ct.err_msg: "should create connection first"}
        self.insert(client, collection_name, data, check_task=CheckTasks.err_res, check_items=error)

    @pytest.mark.tags(CaseLabel.L2)
    def test_milvus_client_insert_column_data(self):
        """
        target: test insert column data
        method: create connection, collection, insert and search
        expected: raise error
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        # 1. create collection
        self.create_collection(client, collection_name, default_dim)
        # 2. insert
        vectors = [[random.random() for _ in range(default_dim)] for _ in range(default_nb)]
        data = [[i for i in range(default_nb)], vectors]
        error = {
            ct.err_code: 999,
            ct.err_msg: "The Input data type is inconsistent with defined schema, please check it.",
        }
        self.insert(client, collection_name, data, check_task=CheckTasks.err_res, check_items=error)
        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L1)
    def test_milvus_client_insert_empty_collection_name(self):
        """
        target: test high level api: client.create_collection
        method: create collection with invalid primary field
        expected: Raise exception
        """
        client = self._client()
        collection_name = ""
        rng = np.random.default_rng(seed=19530)
        rows = [
            {
                default_primary_key_field_name: i,
                default_vector_field_name: list(rng.random((1, default_dim))[0]),
                default_float_field_name: i * 1.0,
                default_string_field_name: str(i),
            }
            for i in range(default_nb)
        ]
        error = {ct.err_code: 1, ct.err_msg: f"`collection_name` value {collection_name} is illegal"}
        self.insert(client, collection_name, rows, check_task=CheckTasks.err_res, check_items=error)

    @pytest.mark.tags(CaseLabel.L2)
    @pytest.mark.parametrize("collection_name", ["12-s", "12 s", "(mn)", "中文", "%$#"])
    def test_milvus_client_insert_invalid_collection_name(self, collection_name):
        """
        target: test high level api: client.create_collection
        method: create collection with invalid primary field
        expected: Raise exception
        """
        client = self._client()
        rng = np.random.default_rng(seed=19530)
        rows = [
            {
                default_primary_key_field_name: i,
                default_vector_field_name: list(rng.random((1, default_dim))[0]),
                default_float_field_name: i * 1.0,
                default_string_field_name: str(i),
            }
            for i in range(default_nb)
        ]
        error = {
            ct.err_code: 1100,
            ct.err_msg: f"Invalid collection name: {collection_name}. the first character of a "
            f"collection name must be an underscore or letter: invalid parameter",
        }
        self.insert(client, collection_name, rows, check_task=CheckTasks.err_res, check_items=error)

    @pytest.mark.tags(CaseLabel.L2)
    def test_milvus_client_insert_collection_name_over_max_length(self):
        """
        target: test high level api: client.create_collection
        method: create collection with invalid primary field
        expected: Raise exception
        """
        client = self._client()
        collection_name = "a".join("a" for i in range(256))
        rng = np.random.default_rng(seed=19530)
        rows = [
            {
                default_primary_key_field_name: i,
                default_vector_field_name: list(rng.random((1, default_dim))[0]),
                default_float_field_name: i * 1.0,
                default_string_field_name: str(i),
            }
            for i in range(default_nb)
        ]
        error = {ct.err_code: 1100, ct.err_msg: f"the length of a collection name must be less than 255 characters"}
        self.insert(client, collection_name, rows, check_task=CheckTasks.err_res, check_items=error)

    @pytest.mark.tags(CaseLabel.L2)
    def test_milvus_client_insert_not_exist_collection_name(self):
        """
        target: test high level api: client.create_collection
        method: create collection with invalid primary field
        expected: Raise exception
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        rng = np.random.default_rng(seed=19530)
        rows = [
            {
                default_primary_key_field_name: i,
                default_vector_field_name: list(rng.random((1, default_dim))[0]),
                default_float_field_name: i * 1.0,
                default_string_field_name: str(i),
            }
            for i in range(default_nb)
        ]
        error = {ct.err_code: 100, ct.err_msg: f"can't find collection[database=default][collection={collection_name}]"}
        self.insert(client, collection_name, rows, check_task=CheckTasks.err_res, check_items=error)

    @pytest.mark.tags(CaseLabel.L2)
    @pytest.mark.parametrize("data", ["12-s", "中文", "%$#", " ", ""])
    def test_milvus_client_insert_data_invalid_type(self, data):
        """
        target: test high level api: client.create_collection
        method: create collection with invalid primary field
        expected: Raise exception
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        # 1. create collection
        self.create_collection(client, collection_name, default_dim, consistency_level="Strong")
        # 2. insert
        error = {
            ct.err_code: 999,
            ct.err_msg: "wrong type of argument 'data',expected 'Dict' or list of 'Dict', got 'str'",
        }
        self.insert(client, collection_name, data, check_task=CheckTasks.err_res, check_items=error)

    @pytest.mark.tags(CaseLabel.L1)
    def test_milvus_client_insert_data_vector_field_missing(self):
        """
        target: test insert entities, with no vector field
        method: vector field is missing in data
        expected: Raise exception
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        # 1. create collection
        self.create_collection(client, collection_name, default_dim, consistency_level="Strong")
        # 2. insert
        rng = np.random.default_rng(seed=19530)
        rows = [
            {default_primary_key_field_name: i, default_float_field_name: i * 1.0, default_string_field_name: str(i)}
            for i in range(default_nb)
        ]
        error = {
            ct.err_code: 1,
            ct.err_msg: f"Insert missed an field `vector` to collection "
            f"without set nullable==true or set default_value",
        }
        self.insert(client, collection_name, data=rows, check_task=CheckTasks.err_res, check_items=error)

    @pytest.mark.tags(CaseLabel.L1)
    def test_milvus_client_insert_data_id_field_missing(self):
        """
        target: test high level api: client.create_collection
        method: create collection with invalid primary field
        expected: Raise exception
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        # 1. create collection
        self.create_collection(client, collection_name, default_dim, consistency_level="Strong")
        # 2. insert
        rng = np.random.default_rng(seed=19530)
        rows = [
            {
                default_vector_field_name: list(rng.random((1, default_dim))[0]),
                default_float_field_name: i * 1.0,
                default_string_field_name: str(i),
            }
            for i in range(default_nb)
        ]
        error = {
            ct.err_code: 1,
            ct.err_msg: f"Insert missed an field `id` to collection without set nullable==true or set default_value",
        }
        self.insert(client, collection_name, data=rows, check_task=CheckTasks.err_res, check_items=error)

    @pytest.mark.tags(CaseLabel.L1)
    def test_milvus_client_insert_data_extra_field(self):
        """
        target: test milvus client: insert extra field than schema
        method: insert extra field than schema when enable_dynamic_field is False
        expected: Raise exception
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        # 1. create collection
        self.create_collection(client, collection_name, default_dim, enable_dynamic_field=False)
        # 2. insert
        rng = np.random.default_rng(seed=19530)
        rows = [
            {
                default_primary_key_field_name: i,
                default_vector_field_name: list(rng.random((1, default_dim))[0]),
                default_float_field_name: i * 1.0,
                default_string_field_name: str(i),
            }
            for i in range(default_nb)
        ]
        error = {
            ct.err_code: 1,
            ct.err_msg: f"Attempt to insert an unexpected field `float` to collection without enabling dynamic field",
        }
        self.insert(client, collection_name, data=rows, check_task=CheckTasks.err_res, check_items=error)

    @pytest.mark.tags(CaseLabel.L1)
    def test_milvus_client_insert_data_dim_not_match(self):
        """
        target: test milvus client: insert extra field than schema
        method: insert extra field than schema when enable_dynamic_field is False
        expected: Raise exception
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        # 1. create collection
        self.create_collection(client, collection_name, default_dim)
        # 2. insert
        rng = np.random.default_rng(seed=19530)
        rows = [
            {
                default_primary_key_field_name: i,
                default_vector_field_name: list(rng.random((1, default_dim + 1))[0]),
                default_float_field_name: i * 1.0,
                default_string_field_name: str(i),
            }
            for i in range(default_nb)
        ]
        error = {ct.err_code: 65536, ct.err_msg: f"of float data should divide the dim({default_dim})"}
        self.insert(client, collection_name, data=rows, check_task=CheckTasks.err_res, check_items=error)

    @pytest.mark.tags(CaseLabel.L2)
    def test_insert_binary_dim_not_match(self):
        """
        target: test insert binary with dim not match
        method: insert binary data dim not equal to schema
        expected: raise exception
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()

        # Create binary vector collection
        schema = self.create_schema(client, enable_dynamic_field=True)[0]
        schema.add_field(default_primary_key_field_name, DataType.INT64, is_primary=True, auto_id=False)
        schema.add_field(ct.default_binary_vec_field_name, DataType.BINARY_VECTOR, dim=default_dim)
        schema.add_field(default_float_field_name, DataType.FLOAT)
        schema.add_field(default_string_field_name, DataType.VARCHAR, max_length=ct.default_length)

        self.create_collection(client, collection_name, dimension=default_dim, schema=schema)

        # Insert binary data
        rng = np.random.default_rng(seed=19530)
        binary_vectors = cf.gen_binary_vectors(num=default_nb, dim=default_dim + 1)[1]
        rows = [
            {
                default_primary_key_field_name: i,
                default_vector_field_name: list(rng.random((1, default_dim + 1))[0]),
                default_float_field_name: i * 1.0,
                default_string_field_name: str(i),
                default_binary_vec_field_name: binary_vectors[i],
            }
            for i in range(default_nb)
        ]

        error = {ct.err_code: 65536, ct.err_msg: f"of all bits should divide the dim({default_dim})"}
        self.insert(client, collection_name, data=rows, check_task=CheckTasks.err_res, check_items=error)

    @pytest.mark.tags(CaseLabel.L1)
    def test_milvus_client_insert_not_matched_data(self):
        """
        target: test milvus client: insert not matched data then defined
        method: insert string to int primary field
        expected: Raise exception
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        # 1. create collection
        self.create_collection(client, collection_name, default_dim)
        # 2. insert
        rng = np.random.default_rng(seed=19530)
        rows = [
            {
                default_primary_key_field_name: str(i),
                default_vector_field_name: list(rng.random((1, default_dim))[0]),
                default_float_field_name: i * 1.0,
                default_string_field_name: str(i),
            }
            for i in range(default_nb)
        ]
        error = {
            ct.err_code: 1,
            ct.err_msg: f"The Input data type is inconsistent with defined schema, {{id}} field should be a int64",
        }
        self.insert(client, collection_name, data=rows, check_task=CheckTasks.err_res, check_items=error)

    @pytest.mark.tags(CaseLabel.L1)
    @pytest.mark.parametrize("partition_name", ["12 s", "(mn)", "中文", "%$#", " "])
    def test_milvus_client_insert_invalid_partition_name(self, partition_name):
        """
        target: test milvus client: insert extra field than schema
        method: insert extra field than schema when enable_dynamic_field is False
        expected: Raise exception
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        # 1. create collection
        self.create_collection(client, collection_name, default_dim)
        # 2. insert
        rng = np.random.default_rng(seed=19530)
        rows = [
            {
                default_primary_key_field_name: i,
                default_vector_field_name: list(rng.random((1, default_dim))[0]),
                default_float_field_name: i * 1.0,
                default_string_field_name: str(i),
            }
            for i in range(default_nb)
        ]
        error = {ct.err_code: 65535, ct.err_msg: f"Invalid partition name: {partition_name}."}
        if partition_name == " ":
            error = {ct.err_code: 1, ct.err_msg: f"Invalid partition name: . Partition name should not be empty."}
        self.insert(
            client,
            collection_name,
            data=rows,
            partition_name=partition_name,
            check_task=CheckTasks.err_res,
            check_items=error,
        )

    @pytest.mark.tags(CaseLabel.L1)
    def test_milvus_client_insert_not_exist_partition_name(self):
        """
        target: test milvus client: insert extra field than schema
        method: insert extra field than schema when enable_dynamic_field is False
        expected: Raise exception
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        # 1. create collection
        self.create_collection(client, collection_name, default_dim)
        # 2. insert
        rng = np.random.default_rng(seed=19530)
        rows = [
            {
                default_primary_key_field_name: i,
                default_vector_field_name: list(rng.random((1, default_dim))[0]),
                default_float_field_name: i * 1.0,
                default_string_field_name: str(i),
            }
            for i in range(default_nb)
        ]
        partition_name = cf.gen_unique_str("partition_not_exist")
        error = {ct.err_code: 200, ct.err_msg: f"partition not found[partition={partition_name}]"}
        self.insert(
            client,
            collection_name,
            data=rows,
            partition_name=partition_name,
            check_task=CheckTasks.err_res,
            check_items=error,
        )

    @pytest.mark.tags(CaseLabel.L2)
    def test_milvus_client_insert_collection_partition_not_match(self):
        """
        target: test milvus client: insert extra field than schema
        method: insert extra field than schema when enable_dynamic_field is False
        expected: Raise exception
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        another_collection_name = cf.gen_collection_name_by_testcase_name()
        partition_name = cf.gen_unique_str("partition")
        # 1. create collection
        self.create_collection(client, collection_name, default_dim)
        self.create_collection(client, another_collection_name, default_dim)
        self.create_partition(client, another_collection_name, partition_name)
        # 2. insert
        rng = np.random.default_rng(seed=19530)
        rows = [
            {
                default_primary_key_field_name: i,
                default_vector_field_name: list(rng.random((1, default_dim))[0]),
                default_float_field_name: i * 1.0,
                default_string_field_name: str(i),
            }
            for i in range(default_nb)
        ]
        error = {ct.err_code: 200, ct.err_msg: f"partition not found[partition={partition_name}]"}
        self.insert(
            client,
            collection_name,
            data=rows,
            partition_name=partition_name,
            check_task=CheckTasks.err_res,
            check_items=error,
        )

    @pytest.mark.tags(CaseLabel.L1)
    def test_insert_ids_binary_invalid(self):
        """
        target: test insert float vector into a collection with binary vector schema
        method: create binary vector collection and insert float vector data
        expected: raise exception
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()

        # 1. Create binary vector collection
        schema = self.create_schema(client, enable_dynamic_field=False)[0]
        schema.add_field(default_primary_key_field_name, DataType.INT64, is_primary=True, auto_id=False)
        schema.add_field(ct.default_binary_vec_field_name, DataType.BINARY_VECTOR, dim=default_dim)
        schema.add_field(default_float_field_name, DataType.FLOAT)
        schema.add_field(default_string_field_name, DataType.VARCHAR, max_length=ct.default_length)

        self.create_collection(client, collection_name, dimension=default_dim, schema=schema)

        # 2. Generate float vector data (invalid for binary vector collection)
        rng = np.random.default_rng(seed=19530)
        rows = [
            {
                default_primary_key_field_name: i,
                default_binary_vec_field_name: list(rng.random((1, default_dim))[0]),
                default_float_field_name: i * 1.0,
                default_string_field_name: str(i),
            }
            for i in range(default_nb)
        ]

        # 3. Verify error on insert
        error = {
            ct.err_code: 999,
            ct.err_msg: "The Input data type is inconsistent with defined schema, {binary_vector} field should be a binary_vector, but got a {<class 'list'>} instead.",
        }
        self.insert(client, collection_name, data=rows, check_task=CheckTasks.err_res, check_items=error)

        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L2)
    def test_insert_with_invalid_binary_partition_name(self):
        """
        target: test insert with invalid scenario
        method: insert binary vector data with invalid partition name
        expected: raise exception
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        partition_name = "non_existent_partition"
        nb = 100

        # 1. Create binary vector collection
        schema = self.create_schema(client, enable_dynamic_field=False)[0]
        schema.add_field(default_primary_key_field_name, DataType.INT64, is_primary=True, auto_id=False)
        schema.add_field(ct.default_binary_vec_field_name, DataType.BINARY_VECTOR, dim=default_dim)
        schema.add_field(default_float_field_name, DataType.FLOAT)
        schema.add_field(default_string_field_name, DataType.VARCHAR, max_length=ct.default_length)

        self.create_collection(client, collection_name, dimension=default_dim, schema=schema)

        # 2. Generate binary vector data
        rows = cf.gen_row_data_by_schema(nb=nb, schema=schema)

        # 3. Verify error on insert with non-existent partition
        error = {ct.err_code: 999, ct.err_msg: f"partition not found[partition={partition_name}]"}
        self.insert(
            client,
            collection_name,
            data=rows,
            partition_name=partition_name,
            check_task=CheckTasks.err_res,
            check_items=error,
        )

        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L2)
    @pytest.mark.parametrize("invalid_int8", [-129, 128])
    def test_insert_int8_overflow(self, invalid_int8):
        """
        target: test insert int8 out of range
        method: insert int8 out of range
        expected: raise exception
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()

        # 1. Create schema with INT8 field
        schema = cf.gen_all_datatype_collection_schema(
            dim=default_dim, enable_dynamic_field=True, enable_struct_array_field=False
        )
        # Add INT8 field
        schema.add_field(ct.default_int8_field_name, DataType.INT8)

        # 2. Create collection
        self.create_collection(client, collection_name, dimension=default_dim, schema=schema)

        # 3. Generate row data
        rows = cf.gen_row_data_by_schema(nb=1, schema=schema)

        # 4. Set invalid INT8 value (out of range: [-128, 127])
        rows[0][ct.default_int8_field_name] = invalid_int8

        # 5. Verify error on insert
        error = {ct.err_code: 1100, ct.err_msg: f"the 0th element ({invalid_int8}) out of range: [-128, 127]"}
        self.insert(client, collection_name, data=rows, check_task=CheckTasks.err_res, check_items=error)

        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L2)
    @pytest.mark.parametrize("invalid_int16", [-32769, 32768])
    def test_insert_int16_overflow(self, invalid_int16):
        """
        target: test insert int16 out of range
        method: insert int16 out of range
        expected: raise exception
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()

        # 1. Create schema with INT16 field
        schema = cf.gen_all_datatype_collection_schema(
            dim=default_dim, enable_dynamic_field=True, enable_struct_array_field=False
        )
        # Add INT16 field
        schema.add_field(ct.default_int16_field_name, DataType.INT16)

        # 2. Create collection
        self.create_collection(client, collection_name, dimension=default_dim, schema=schema)

        # 3. Generate row data
        rows = cf.gen_row_data_by_schema(nb=1, schema=schema)

        # 4. Set invalid INT16 value (out of range: [-32768, 32767])
        rows[0][ct.default_int16_field_name] = invalid_int16

        # 5. Verify error on insert
        error = {ct.err_code: 1100, ct.err_msg: f"the 0th element ({invalid_int16}) out of range: [-32768, 32767]"}
        self.insert(client, collection_name, data=rows, check_task=CheckTasks.err_res, check_items=error)

        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L2)
    @pytest.mark.parametrize("invalid_int32", [-2147483649, 2147483648])
    def test_insert_int32_overflow(self, invalid_int32):
        """
        target: test insert int32 out of range
        method: insert int32 out of range
        expected: raise exception
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()

        # 1. Create schema with INT32 field
        schema = cf.gen_all_datatype_collection_schema(
            dim=default_dim, enable_dynamic_field=True, enable_struct_array_field=False
        )
        # Add INT32 field
        schema.add_field(ct.default_int32_field_name, DataType.INT32)

        # 2. Create collection
        self.create_collection(client, collection_name, dimension=default_dim, schema=schema)

        # 3. Generate row data
        rows = cf.gen_row_data_by_schema(nb=1, schema=schema)

        # 4. Set invalid INT32 value (out of range: [-2147483648, 2147483647])
        rows[0][ct.default_int32_field_name] = invalid_int32

        # 5. Verify error on insert
        error = {ct.err_code: 1, ct.err_msg: "The Input data type is inconsistent with defined schema"}
        self.insert(client, collection_name, data=rows, check_task=CheckTasks.err_res, check_items=error)

        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L0)
    @pytest.mark.parametrize("primary_field", [ct.default_int64_field_name, ct.default_string_field_name])
    def test_insert_with_invalid_field_value(self, primary_field):
        """
        target: verify error msg when inserting with invalid field value
        method: insert with invalid field value
        expected: raise exception
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        nb = 100

        # 1. Create schema
        schema = self.create_schema(client)[0]
        if primary_field == ct.default_int64_field_name:
            schema.add_field(primary_field, DataType.INT64, is_primary=True, auto_id=True)
        else:
            schema.add_field(
                primary_field, DataType.VARCHAR, max_length=ct.default_length, is_primary=True, auto_id=True
            )
        schema.add_field(default_vector_field_name, DataType.FLOAT_VECTOR, dim=default_dim)
        schema.add_field(default_float_field_name, DataType.FLOAT)
        schema.add_field(default_bool_field_name, DataType.BOOL)
        if primary_field != ct.default_string_field_name:
            schema.add_field(default_string_field_name, DataType.VARCHAR, max_length=ct.default_length)

        # 2. Create collection
        self.create_collection(client, collection_name, dimension=default_dim, schema=schema)

        # 3. Generate row data
        rows = cf.gen_row_data_by_schema(nb=nb, schema=schema)

        # 4. Test invalid field values at different positions
        for dirty_i in [0, nb // 2, nb - 1]:  # check the dirty data at first, middle and last
            # Iterate through all fields in the row
            for field_name, field_value in rows[dirty_i].items():
                # Get the actual value type
                value_type = type(field_value)
                error = {ct.err_code: 999, ct.err_msg: "The Input data type is inconsistent with defined schema"}

                # Inject type errors based on value type (only for simple scalar types)
                if value_type in (int, bool, float):
                    tmp = rows[dirty_i][field_name]
                    rows[dirty_i][field_name] = "iamstring"
                    self.insert(client, collection_name, data=rows, check_task=CheckTasks.err_res, check_items=error)
                    rows[dirty_i][field_name] = tmp
                elif value_type is str:
                    tmp = rows[dirty_i][field_name]
                    rows[dirty_i][field_name] = random.randint(0, 1000)
                    self.insert(client, collection_name, data=rows, check_task=CheckTasks.err_res, check_items=error)
                    rows[dirty_i][field_name] = tmp
                else:
                    continue

        # 5. Verify correct data can be inserted
        results = self.insert(client, collection_name, rows)[0]
        assert results["insert_count"] == nb

        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L2)
    def test_insert_over_resource_limit(self):
        """
        target: test insert over RPC limitation 64MB (67108864)
        method: insert excessive data
        expected: raise exception
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        nb = 150000

        # 1. Create collection
        self.create_collection(client, collection_name, default_dim, auto_id=False)

        # 2. Generate row data (150000 rows, which exceeds 64MB RPC limit)
        rng = np.random.default_rng(seed=19530)
        rows = [
            {
                default_primary_key_field_name: i,
                default_vector_field_name: list(rng.random((1, default_dim))[0]),
                default_float_field_name: i * 1.0,
                default_string_field_name: str(i),
            }
            for i in range(nb)
        ]

        # 3. Verify error on insert
        error = {ct.err_code: 999, ct.err_msg: "message larger than max"}
        self.insert(client, collection_name, data=rows, check_task=CheckTasks.err_res, check_items=error)

        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L2)
    @pytest.mark.parametrize("default_value", [[], 123])
    def test_insert_type_mismatch_with_default_value_field(self, default_value):
        """
        target: test insert with type mismatch for field that has default value
        method: insert data with wrong type for varchar field that has default_value
        expected: raise exception
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()

        # 1. Create schema with default value field
        schema = self.create_schema(client, enable_dynamic_field=False)[0]
        schema.add_field(default_primary_key_field_name, DataType.INT64, is_primary=True, auto_id=False)
        schema.add_field(default_float_field_name, DataType.FLOAT)
        schema.add_field(default_string_field_name, DataType.VARCHAR, max_length=ct.default_length, default_value="abc")
        schema.add_field(default_vector_field_name, DataType.FLOAT_VECTOR, dim=default_dim)

        # 2. Create collection
        self.create_collection(client, collection_name, dimension=default_dim, schema=schema)

        # 3. Generate vectors
        vectors = cf.gen_vectors(ct.default_nb, ct.default_dim)

        # 4. Prepare test data with invalid type for varchar field
        data = [
            {
                default_primary_key_field_name: 1,
                default_float_field_name: 1.0,
                default_string_field_name: default_value,
                default_vector_field_name: vectors[0],
            }
        ]

        # 5. Verify error on upsert
        error = {ct.err_code: 999, ct.err_msg: "The Input data type is inconsistent with defined schema"}
        self.upsert(client, collection_name, data=data, check_task=CheckTasks.err_res, check_items=error)

        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L2)
    def test_insert_with_nan_value(self):
        """
        target: test insert with nan value
        method: insert with nan value: None, float('nan'), np.NAN/np.nan, float('inf')
        expected: raise exception
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()

        # 1. Create collection
        self.create_collection(client, collection_name, default_dim, auto_id=False)

        # 2. Generate row data
        rng = np.random.default_rng(seed=19530)
        rows = [
            {
                default_primary_key_field_name: i,
                default_vector_field_name: list(rng.random((1, default_dim))[0]),
                default_float_field_name: i * 1.0,
                default_string_field_name: str(i),
            }
            for i in range(default_nb)
        ]

        # 3. Test None value in vector field
        rows[0][default_vector_field_name][0] = None
        error = {ct.err_code: 999, ct.err_msg: "The Input data type is inconsistent with defined schema"}
        self.insert(client, collection_name, data=rows, check_task=CheckTasks.err_res, check_items=error)

        # 4. Test float('nan') in vector field
        rows[0][default_vector_field_name][0] = float("nan")
        error = {ct.err_code: 999, ct.err_msg: "value 'NaN' is not a number or infinity"}
        self.insert(client, collection_name, data=rows, check_task=CheckTasks.err_res, check_items=error)

        # 5. Test np.NAN in vector field
        rows[0][default_vector_field_name][0] = np.nan
        self.insert(client, collection_name, data=rows, check_task=CheckTasks.err_res, check_items=error)

        # 6. Test float('inf') in vector field
        rows[0][default_vector_field_name][0] = float("inf")
        error = {ct.err_code: 65535, ct.err_msg: "value '+Inf' is not a number or infinity"}
        self.insert(client, collection_name, data=rows, check_task=CheckTasks.err_res, check_items=error)

        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L2)
    @pytest.mark.parametrize("invalid_vector_type", ct.all_dense_vector_types)
    def test_invalid_sparse_vector_data(self, invalid_vector_type):
        """
        target: insert illegal data type
        method: insert illegal dense vector type into sparse vector field
        expected: raise exception
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        nb = 100

        # 1. Create schema with sparse vector field
        schema = self.create_schema(client, enable_dynamic_field=False)[0]
        schema.add_field(ct.default_int64_field_name, DataType.INT64, is_primary=True, auto_id=False)
        schema.add_field(ct.default_float_field_name, DataType.FLOAT)
        schema.add_field(ct.default_string_field_name, DataType.VARCHAR, max_length=ct.default_length)
        schema.add_field(ct.default_sparse_vec_field_name, DataType.SPARSE_FLOAT_VECTOR)

        # 2. Create collection
        self.create_collection(client, collection_name, dimension=default_dim, schema=schema)

        # 3. Generate valid sparse vector data
        sparse_vectors = cf.gen_sparse_vectors(nb, dim=128)
        rows = []
        for i in range(nb - 1):
            row = {
                ct.default_int64_field_name: i,
                ct.default_float_field_name: np.float32(i),
                ct.default_string_field_name: str(i),
                ct.default_sparse_vec_field_name: sparse_vectors[i],
            }
            rows.append(row)

        # 4. Add invalid dense vector type as the last row
        invalid_vec = cf.gen_vectors(1, dim=128, vector_data_type=invalid_vector_type)
        invalid_row = {
            ct.default_int64_field_name: nb - 1,
            ct.default_float_field_name: np.float32(nb - 1),
            ct.default_string_field_name: str(nb - 1),
            ct.default_sparse_vec_field_name: invalid_vec[0],
        }
        rows.append(invalid_row)

        # 5. Verify error on insert
        error = {ct.err_code: 1, ct.err_msg: "invalid input for sparse float vector"}
        self.insert(client, collection_name, data=rows, check_task=CheckTasks.err_res, check_items=error)

        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L2)
    def test_insert_with_pk_varchar_auto_id_true(self):
        """
        target: test insert with pk varchar and auto id true
        method: set pk varchar max length < 18, insert data
        expected: varchar pk supports auto_id=true
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        rng = np.random.default_rng()

        # 1. Create schema with varchar pk (max_length=6) and auto_id=True
        schema = self.create_schema(client, enable_dynamic_field=False)[0]
        schema.add_field(ct.default_string_field_name, DataType.VARCHAR, max_length=6, is_primary=True, auto_id=True)
        schema.add_field(default_vector_field_name, DataType.FLOAT_VECTOR, dim=default_dim)

        # 2. Create collection
        self.create_collection(client, collection_name, schema=schema)

        # 3. Insert 2 rows (no pk field since auto_id=True)
        rows = [{default_vector_field_name: list(rng.random(default_dim).astype(np.float32))} for _ in range(2)]
        res = self.insert(client, collection_name, data=rows)[0]
        assert res["insert_count"] == 2

        # 4. Create index and load
        index_params = self.prepare_index_params(client)[0]
        index_params.add_index(field_name=default_vector_field_name, index_type="FLAT", metric_type="L2")
        self.create_index(client, collection_name, index_params)
        self.load_collection(client, collection_name)

        # 5. Query to verify inserted data has auto-generated ids
        results = self.query(client, collection_name, filter="", limit=10)[0]
        assert len(results) == 2
        for r in results:
            assert ct.default_string_field_name in r
            assert len(r[ct.default_string_field_name]) > 0

        self.drop_collection(client, collection_name)


class TestMilvusClientInsertValid(TestMilvusClientV2Base):
    """Test case of search interface"""

    @pytest.fixture(scope="function", params=[False, True])
    def auto_id(self, request):
        yield request.param

    @pytest.fixture(scope="function", params=["COSINE", "L2"])
    def metric_type(self, request):
        yield request.param

    @pytest.fixture(scope="function", params=[True, False])
    def nullable(self, request):
        yield request.param

    @pytest.fixture(
        scope="function",
        params=[DataType.FLOAT_VECTOR, DataType.FLOAT16_VECTOR, DataType.BFLOAT16_VECTOR, DataType.INT8_VECTOR],
    )
    def vector_type(self, request):
        yield request.param

    """
    ******************************************************************
    #  The following are valid base cases
    ******************************************************************
    """

    @pytest.mark.tags(CaseLabel.L0)
    def test_milvus_client_insert_default(self, vector_type, nullable):
        """
        target: test search (high level api) normal case
        method: create connection, collection, insert and search
        expected: search/query successfully
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        # 1. create collection
        dim = 8
        # 1. create collection
        schema = self.create_schema(client, enable_dynamic_field=False)[0]
        schema.add_field(default_primary_key_field_name, DataType.INT64, max_length=64, is_primary=True, auto_id=False)
        schema.add_field(default_vector_field_name, vector_type, dim=dim, nullable=nullable)
        schema.add_field(default_string_field_name, DataType.VARCHAR, max_length=64, is_partition_key=True)
        schema.add_field(default_float_field_name, DataType.FLOAT, nullable=nullable)
        index_params = self.prepare_index_params(client)[0]
        index_params.add_index(default_vector_field_name, metric_type="COSINE")
        self.create_collection(client, collection_name, dimension=dim, schema=schema, index_params=index_params)
        # 2. insert
        rows = cf.gen_row_data_by_schema(ct.default_nb, schema=schema)
        results = self.insert(client, collection_name, rows)[0]
        assert results["insert_count"] == default_nb
        # 3. search
        vectors_to_search = cf.gen_vectors(ct.default_nq, dim=dim, vector_data_type=vector_type)
        insert_ids = [i for i in range(default_nb)]
        self.search(
            client,
            collection_name,
            vectors_to_search,
            check_task=CheckTasks.check_search_results,
            check_items={
                "enable_milvus_client_api": True,
                "nq": len(vectors_to_search),
                "ids": insert_ids,
                "limit": default_limit,
                "pk_name": default_primary_key_field_name,
            },
        )
        # 4. query
        self.query(
            client,
            collection_name,
            filter=default_search_exp,
            check_task=CheckTasks.check_query_results,
            check_items={
                exp_res: rows,
                "with_vec": True,
                "pk_name": default_primary_key_field_name,
                "vector_type": vector_type,
            },
        )
        self.release_collection(client, collection_name)
        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L2)
    def test_milvus_client_insert_binary_default(self):
        """
        target: test insert binary data, test binary vector insert/search using client api
        method: create collection, insert, search and query
        expected: insert/search/query successfully
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()

        # Create binary vector collection
        schema = self.create_schema(client, enable_dynamic_field=True)[0]
        schema.add_field(default_primary_key_field_name, DataType.INT64, is_primary=True, auto_id=False)
        schema.add_field(ct.default_binary_vec_field_name, DataType.BINARY_VECTOR, dim=default_dim)
        schema.add_field(default_float_field_name, DataType.FLOAT)
        schema.add_field(default_string_field_name, DataType.VARCHAR, max_length=ct.default_length)

        index_params = self.prepare_index_params(client)[0]
        index_params.add_index(ct.default_binary_vec_field_name, index_type="BIN_IVF_FLAT", metric_type="HAMMING")

        self.create_collection(client, collection_name, dimension=default_dim, schema=schema, index_params=index_params)

        indexes = self.list_indexes(client, collection_name)[0]
        assert ct.default_binary_vec_field_name in indexes

        # Insert binary data
        rows = cf.gen_row_data_by_schema(nb=ct.default_nb, schema=schema)
        results = self.insert(client, collection_name, rows)[0]
        assert results["insert_count"] == default_nb

        self.flush(client, collection_name)
        num_entities = self.get_collection_stats(client, collection_name)[0]
        assert num_entities.get("row_count", None) == ct.default_nb

        # search
        vectors_to_search = [rows[0][default_binary_vec_field_name]]
        insert_ids = [i for i in range(default_nb)]
        self.search(
            client,
            collection_name,
            vectors_to_search,
            check_task=CheckTasks.check_search_results,
            check_items={
                "enable_milvus_client_api": True,
                "nq": len(vectors_to_search),
                "ids": insert_ids,
                "limit": default_limit,
                "pk_name": default_primary_key_field_name,
            },
        )
        # query
        self.query(
            client,
            collection_name,
            filter=default_search_exp,
            check_task=CheckTasks.check_query_results,
            check_items={"exp_limit": default_nb, "with_vec": False, "vector_type": DataType.BINARY_VECTOR},
        )
        self.release_collection(client, collection_name)
        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L2)
    def test_milvus_client_insert_different_fields(self):
        """
        target: test search (high level api) normal case
        method: create connection, collection, insert and search
        expected: search/query successfully
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        # 1. create collection
        self.create_collection(client, collection_name, default_dim, consistency_level="Strong")
        collections = self.list_collections(client)[0]
        assert collection_name in collections
        self.describe_collection(
            client,
            collection_name,
            check_task=CheckTasks.check_describe_collection_property,
            check_items={"collection_name": collection_name, "dim": default_dim, "consistency_level": 0},
        )
        # 2. insert
        rng = np.random.default_rng(seed=19530)
        rows = [
            {
                default_primary_key_field_name: i,
                default_vector_field_name: list(rng.random((1, default_dim))[0]),
                default_float_field_name: i * 1.0,
                default_string_field_name: str(i),
            }
            for i in range(default_nb)
        ]
        results = self.insert(client, collection_name, rows)[0]
        assert results["insert_count"] == default_nb
        # 3. insert diff fields
        rows = [
            {
                default_primary_key_field_name: i + default_nb,
                default_vector_field_name: list(rng.random((1, default_dim))[0]),
                default_float_field_name: i * 1.0,
                "new_diff_str_field": str(i),
            }
            for i in range(default_nb)
        ]
        results = self.insert(client, collection_name, rows)[0]
        assert results["insert_count"] == default_nb
        # 3. search
        vectors_to_search = rng.random((1, default_dim))
        insert_ids = [i for i in range(default_nb * 2)]
        self.search(
            client,
            collection_name,
            vectors_to_search,
            check_task=CheckTasks.check_search_results,
            check_items={
                "enable_milvus_client_api": True,
                "nq": len(vectors_to_search),
                "ids": insert_ids,
                "limit": default_limit,
                "pk_name": default_primary_key_field_name,
            },
        )
        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L2)
    def test_milvus_client_insert_empty_data(self):
        """
        target: test search (high level api) normal case
        method: create connection, collection, insert and search
        expected: search/query successfully
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        # 1. create collection
        self.create_collection(client, collection_name, default_dim, consistency_level="Strong")
        # 2. insert
        rows = []
        results = self.insert(client, collection_name, rows)[0]
        assert results["insert_count"] == 0
        # 3. search
        rng = np.random.default_rng(seed=19530)
        vectors_to_search = rng.random((1, default_dim))
        self.search(
            client,
            collection_name,
            vectors_to_search,
            check_task=CheckTasks.check_search_results,
            check_items={
                "enable_milvus_client_api": True,
                "nq": len(vectors_to_search),
                "ids": [],
                "pk_name": default_primary_key_field_name,
                "limit": 0,
            },
        )
        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L2)
    def test_insert_with_non_data_type(self):
        """
        target: test insert with none type data
        method: create collection
        expected: milvus client does not support insert with none type data
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()

        # Create collection
        self.create_collection(client, collection_name, dimension=default_dim)

        # Try to insert with none type data
        error = {
            ct.err_code: -1,
            ct.err_msg: f"wrong type of argument 'data',expected 'Dict' or list of 'Dict', got 'NoneType'",
        }
        self.insert(client, collection_name, None, check_task=CheckTasks.err_res, check_items=error)

    @pytest.mark.tags(CaseLabel.L1)
    def test_milvus_client_insert_partition(self):
        """
        target: test fast create collection normal case
        method: create collection
        expected: create collection with default schema, index, and load successfully
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        partition_name = cf.gen_unique_str("partition")
        # 1. create collection
        self.create_collection(client, collection_name, default_dim, consistency_level="Strong")
        # 2. create partition
        self.create_partition(client, collection_name, partition_name)
        partitions = self.list_partitions(client, collection_name)[0]
        assert partition_name in partitions
        index = self.list_indexes(client, collection_name)[0]
        assert index == ["vector"]
        # load_state = self.get_load_state(collection_name)[0]
        # 3. insert
        rng = np.random.default_rng(seed=19530)
        rows = [
            {
                default_primary_key_field_name: i,
                default_vector_field_name: list(rng.random((1, default_dim))[0]),
                default_float_field_name: i * 1.0,
                default_string_field_name: str(i),
            }
            for i in range(default_nb)
        ]
        results = self.insert(client, collection_name, rows, partition_name=partition_name)[0]
        assert results["insert_count"] == default_nb
        # 3. search
        vectors_to_search = rng.random((1, default_dim))
        insert_ids = [i for i in range(default_nb)]
        self.search(
            client,
            collection_name,
            vectors_to_search,
            check_task=CheckTasks.check_search_results,
            check_items={
                "enable_milvus_client_api": True,
                "nq": len(vectors_to_search),
                "ids": insert_ids,
                "limit": default_limit,
                "pk_name": default_primary_key_field_name,
            },
        )
        # partition_number = self.get_partition_stats(client, collection_name, "_default")[0]
        # assert partition_number == default_nb
        # partition_number = self.get_partition_stats(client, collection_name, partition_name)[0]
        # assert partition_number[0]['value'] == 0
        if self.has_partition(client, collection_name, partition_name)[0]:
            self.release_partitions(client, collection_name, partition_name)
            self.drop_partition(client, collection_name, partition_name)
        if self.has_collection(client, collection_name)[0]:
            self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L0)
    def test_insert_binary_partition(self):
        """
        target: test insert entities and create partition
        method: create collection and insert binary entities in it, with the partition_name param
        expected: the collection row count equals to nb
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        partition_name = cf.gen_unique_str("partition")

        # 1. Create binary vector collection
        schema = self.create_schema(client, enable_dynamic_field=False)[0]
        schema.add_field(default_primary_key_field_name, DataType.INT64, is_primary=True, auto_id=False)
        schema.add_field(ct.default_binary_vec_field_name, DataType.BINARY_VECTOR, dim=default_dim)
        schema.add_field(default_float_field_name, DataType.FLOAT)
        schema.add_field(default_string_field_name, DataType.VARCHAR, max_length=ct.default_length)

        self.create_collection(client, collection_name, dimension=default_dim, schema=schema)

        # 2. Create partition
        self.create_partition(client, collection_name, partition_name)
        partitions = self.list_partitions(client, collection_name)[0]
        assert partition_name in partitions

        # 3. Insert binary data into partition
        rows = cf.gen_row_data_by_schema(nb=ct.default_nb, schema=schema)
        results = self.insert(client, collection_name, rows, partition_name=partition_name)[0]
        assert results["insert_count"] == ct.default_nb

        # 4. Verify row count
        self.flush(client, collection_name)
        num_entities = self.get_collection_stats(client, collection_name)[0]
        assert num_entities.get("row_count", None) == ct.default_nb

        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L2)
    @pytest.mark.parametrize("default_value", ["a" * 64, "aa"])
    def test_milvus_client_insert_with_added_field(self, default_value):
        """
        target: test search (high level api) normal case
        method: create connection, collection, insert, add field, insert and search
        expected: search/query successfully
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        dim = 8
        # 1. create collection
        schema = self.create_schema(client, enable_dynamic_field=False)[0]
        schema.add_field(default_primary_key_field_name, DataType.INT64, max_length=64, is_primary=True, auto_id=False)
        schema.add_field(default_vector_field_name, DataType.FLOAT_VECTOR, dim=dim)
        schema.add_field(default_string_field_name, DataType.VARCHAR, max_length=64, is_partition_key=True)
        schema.add_field(default_float_field_name, DataType.FLOAT, nullable=True)
        index_params = self.prepare_index_params(client)[0]
        index_params.add_index(default_vector_field_name, metric_type="COSINE")
        self.create_collection(client, collection_name, dimension=dim, schema=schema, index_params=index_params)
        # 2. insert before add field
        vectors = cf.gen_vectors(default_nb * 2, dim, vector_data_type=DataType.FLOAT_VECTOR)
        rows = [
            {
                default_primary_key_field_name: i,
                default_vector_field_name: vectors[i],
                default_float_field_name: i * 1.0,
                default_string_field_name: str(i),
            }
            for i in range(default_nb)
        ]
        results = self.insert(client, collection_name, rows)[0]
        assert results["insert_count"] == default_nb
        # 3. add new field
        self.add_collection_field(
            client,
            collection_name,
            field_name="field_new",
            data_type=DataType.VARCHAR,
            nullable=True,
            default_value=default_value,
            max_length=64,
        )
        vectors_to_search = [vectors[0]]
        insert_ids = [i for i in range(default_nb)]
        # 4. check old dynamic data search is not impacted after add new field
        self.search(
            client,
            collection_name,
            vectors_to_search,
            check_task=CheckTasks.check_search_results,
            check_items={
                "enable_milvus_client_api": True,
                "nq": len(vectors_to_search),
                "ids": insert_ids,
                "pk_name": default_primary_key_field_name,
                "limit": default_limit,
            },
        )
        # 5. insert data(old + new field)
        rows_t = [
            {
                default_primary_key_field_name: i,
                default_vector_field_name: vectors[i],
                default_float_field_name: i * 1.0,
                default_string_field_name: str(i),
                "field_new": "field_new",
            }
            for i in range(default_nb, default_nb * 2)
        ]
        results = self.insert(client, collection_name, rows_t)[0]
        assert results["insert_count"] == default_nb
        insert_ids_after_add_field = [i for i in range(default_nb, default_nb * 2)]
        # 6. search filtered with the new field
        self.search(
            client,
            collection_name,
            vectors_to_search,
            filter=f'field_new=="{default_value}"',
            check_task=CheckTasks.check_search_results,
            check_items={
                "enable_milvus_client_api": True,
                "nq": len(vectors_to_search),
                "ids": insert_ids,
                "pk_name": default_primary_key_field_name,
                "limit": default_limit,
            },
        )
        self.search(
            client,
            collection_name,
            vectors_to_search,
            filter=f"field_new=='field_new'",
            check_task=CheckTasks.check_search_results,
            check_items={
                "enable_milvus_client_api": True,
                "nq": len(vectors_to_search),
                "ids": insert_ids_after_add_field,
                "pk_name": default_primary_key_field_name,
                "limit": default_limit,
            },
        )
        self.release_collection(client, collection_name)
        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L0)
    @pytest.mark.parametrize("nb", [1, default_nb])
    def test_insert_row_data(self, nb):
        """
        target: test insert row-based data
        method: 1.create collection with explicit schema 2.insert row data
        expected: assert num entities
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        # 1. create collection with explicit schema
        schema = self.create_schema(client, enable_dynamic_field=False)[0]
        schema.add_field(default_primary_key_field_name, DataType.INT64, is_primary=True, auto_id=False)
        schema.add_field(default_vector_field_name, DataType.FLOAT_VECTOR, dim=default_dim)
        schema.add_field(default_float_field_name, DataType.FLOAT)
        schema.add_field(default_string_field_name, DataType.VARCHAR, max_length=ct.default_length)
        self.create_collection(client, collection_name, schema=schema)
        # 2. insert
        rng = np.random.default_rng(seed=19530)
        rows = [
            {
                default_primary_key_field_name: i,
                default_vector_field_name: list(rng.random((1, default_dim))[0]),
                default_float_field_name: i * 1.0,
                default_string_field_name: str(i),
            }
            for i in range(nb)
        ]
        results = self.insert(client, collection_name, rows)[0]
        assert results["insert_count"] == nb
        assert sorted(results["ids"]) == list(range(nb))
        # 3. verify num entities
        self.flush(client, collection_name)
        stats = self.get_collection_stats(client, collection_name)[0]
        assert stats.get("row_count", None) == nb
        self.drop_collection(client, collection_name)


class TestInsertOperation(TestMilvusClientV2Base):
    """
    ******************************************************************
      The following cases are used to test insert interface operations
    ******************************************************************
    """

    @pytest.fixture(scope="function", params=[8, 4096])
    def dim(self, request):
        yield request.param

    @pytest.fixture(scope="function", params=[False, True])
    def auto_id(self, request):
        yield request.param

    @pytest.fixture(scope="function", params=[ct.default_int64_field_name, ct.default_string_field_name])
    def pk_field(self, request):
        yield request.param

    @pytest.mark.tags(CaseLabel.L2)
    def test_insert_without_connection(self):
        """
        target: test insert without connection
        method: insert after remove connection
        expected: raise exception
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        self.create_collection(client, collection_name, default_dim)
        self.close(client)

        rng = np.random.default_rng(seed=19530)
        rows = [
            {
                default_primary_key_field_name: i,
                default_vector_field_name: list(rng.random((1, default_dim))[0]),
                default_float_field_name: i * 1.0,
                default_string_field_name: str(i),
            }
            for i in range(10)
        ]
        error = {ct.err_code: 999, ct.err_msg: "should create connection first"}
        self.insert(client, collection_name, rows, check_task=CheckTasks.err_res, check_items=error)

    @pytest.mark.skip(reason="Covered by test_milvus_client_insert_partition ")
    @pytest.mark.tags(CaseLabel.L1)
    def test_insert_default_partition(self):
        """
        target: test insert entities into default partition
        method: create partition and insert info collection
        expected: the collection insert count equals to nb
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        partition_name = cf.gen_unique_str("partition")

        self.create_collection(client, collection_name, default_dim)
        self.create_partition(client, collection_name, partition_name)

        rng = np.random.default_rng(seed=19530)
        rows = [
            {
                default_primary_key_field_name: i,
                default_vector_field_name: list(rng.random((1, default_dim))[0]),
                default_float_field_name: i * 1.0,
                default_string_field_name: str(i),
            }
            for i in range(ct.default_nb)
        ]
        results = self.insert(client, collection_name, rows, partition_name=partition_name)[0]
        assert results["insert_count"] == ct.default_nb
        self.drop_collection(client, collection_name)

    @pytest.mark.skip(reason="Covered by test_milvus_client_insert_not_exist_partition_name ")
    def test_insert_partition_not_existed(self):
        """
        target: test insert entities in collection created before
        method: create collection and insert entities in it, with the not existed partition_name param
        expected: error raised
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        self.create_collection(client, collection_name, default_dim)

        rng = np.random.default_rng(seed=19530)
        rows = [
            {
                default_primary_key_field_name: i,
                default_vector_field_name: list(rng.random((1, default_dim))[0]),
                default_float_field_name: i * 1.0,
                default_string_field_name: str(i),
            }
            for i in range(10)
        ]
        error = {ct.err_code: 200, ct.err_msg: "partition not found[partition=p]"}
        self.insert(client, collection_name, rows, partition_name="p", check_task=CheckTasks.err_res, check_items=error)
        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L1)
    def test_insert_partition_repeatedly(self):
        """
        target: test insert entities in collection created before
        method: create collection and insert entities in it repeatedly, with the partition_name param
        expected: the collection row count equals to nq
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        partition_name_1 = cf.gen_unique_str("partition1")
        partition_name_2 = cf.gen_unique_str("partition2")

        self.create_collection(client, collection_name, default_dim)
        self.create_partition(client, collection_name, partition_name_1)
        self.create_partition(client, collection_name, partition_name_2)

        rng = np.random.default_rng(seed=19530)
        rows = [
            {
                default_primary_key_field_name: i,
                default_vector_field_name: list(rng.random((1, default_dim))[0]),
                default_float_field_name: i * 1.0,
                default_string_field_name: str(i),
            }
            for i in range(ct.default_nb)
        ]
        result_1 = self.insert(client, collection_name, rows, partition_name=partition_name_1)[0]
        result_2 = self.insert(client, collection_name, rows, partition_name=partition_name_2)[0]
        assert result_1["insert_count"] == ct.default_nb
        assert result_2["insert_count"] == ct.default_nb

        self.flush(client, collection_name)
        num_entities = self.get_collection_stats(client, collection_name)[0]
        assert num_entities.get("row_count", None) == 2 * ct.default_nb

        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L0)
    def test_insert_partition_with_ids(self):
        """
        target: test insert entities in collection created before, insert with ids
        method: create collection and insert entities in it, with the partition_name param
        expected: the collection insert count equals to nq
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        partition_name = cf.gen_unique_str("partition")

        self.create_collection(client, collection_name, default_dim)
        self.create_partition(client, collection_name, partition_name)

        rng = np.random.default_rng(seed=19530)
        rows = [
            {
                default_primary_key_field_name: i,
                default_vector_field_name: list(rng.random((1, default_dim))[0]),
                default_float_field_name: i * 1.0,
                default_string_field_name: str(i),
            }
            for i in range(ct.default_nb)
        ]
        results = self.insert(client, collection_name, rows, partition_name=partition_name)[0]
        assert results["insert_count"] == ct.default_nb
        assert sorted(results["ids"]) == list(range(ct.default_nb))
        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L1)
    def test_insert_exceed_varchar_limit(self):
        """
        target: test insert exceed varchar limit
        method: create a collection with varchar limit=2 and insert invalid data
        expected: error raised
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()

        # Create schema with varchar limit
        schema = self.create_schema(client, auto_id=True, enable_dynamic_field=False)[0]
        schema.add_field("id", DataType.INT64, is_primary=True, auto_id=True)
        schema.add_field("vector", DataType.FLOAT_VECTOR, dim=ct.default_dim)
        schema.add_field("small_limit", DataType.VARCHAR, max_length=2)
        schema.add_field("big_limit", DataType.VARCHAR, max_length=65530)

        self.create_collection(client, collection_name, dimension=ct.default_dim, schema=schema)

        # Insert data exceeding varchar limit
        rows = [
            {
                "vector": list(cf.gen_vectors(1, ct.default_dim)[0]),
                "small_limit": "limit_1___________",
                "big_limit": "1",
            },
            {
                "vector": list(cf.gen_vectors(1, ct.default_dim)[0]),
                "small_limit": "limit_2___________",
                "big_limit": "2",
            },
        ]
        error = {ct.err_code: 999, ct.err_msg: "length of varchar field small_limit exceeds max length"}
        self.insert(client, collection_name, rows, check_task=CheckTasks.err_res, check_items=error)
        self.drop_collection(client, collection_name)

    @pytest.mark.skip(reason="duplicate with test_milvus_client_insert_data_vector_field_missing")
    @pytest.mark.tags(CaseLabel.L2)
    def test_insert_with_no_vector_field_dtype(self):
        """
        target: test insert entities, with no vector field
        method: vector field is missing in data
        expected: error raised
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        self.create_collection(client, collection_name, default_dim)

        # Generate data without vector field
        rng = np.random.default_rng(seed=19530)
        rows = [
            {default_primary_key_field_name: i, default_float_field_name: i * 1.0, default_string_field_name: str(i)}
            for i in range(10)
        ]
        error = {ct.err_code: 1, ct.err_msg: f"Insert missed an field `vector` to collection"}
        self.insert(client, collection_name, rows, check_task=CheckTasks.err_res, check_items=error)
        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L1)
    def test_milvus_client_insert_data_field_name_not_match(self):
        """
        target: test insert field name not match
        method: data field name not match schema
        expected: Raise exception
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        # 1. create collection
        self.create_collection(
            client, collection_name, default_dim, consistency_level="Strong", enable_dynamic_field=False
        )
        # 2. insert with wrong field name
        rng = np.random.default_rng(seed=19530)
        rows = [
            {
                default_primary_key_field_name: i,
                "wrong_vector": list(rng.random((1, default_dim))[0]),
            }
            for i in range(default_nb)
        ]
        error = {
            ct.err_code: 1,
            ct.err_msg: f"Attempt to insert an unexpected field `wrong_vector` to collection without enabling dynamic field",
        }
        self.insert(client, collection_name, data=rows, check_task=CheckTasks.err_res, check_items=error)
        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L2)
    def test_insert_with_vector_field_dismatch_dtype(self):
        """
        target: test insert entities with mismatched vector field data type
        method: provide vector field with scalar value instead of list/array
        expected: raise exception due to schema dtype mismatch
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        self.create_collection(client, collection_name, default_dim)

        # Generate data with wrong vector type (scalar instead of list)
        rows = [
            {
                default_primary_key_field_name: 0,
                default_vector_field_name: 0.0001,
                default_float_field_name: 0.0,
                default_string_field_name: "0",
            }
        ]
        error = {ct.err_code: 1, ct.err_msg: "The Input data type is inconsistent with defined schema"}
        self.insert(client, collection_name, rows, check_task=CheckTasks.err_res, check_items=error)
        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L1)
    def test_insert_drop_collection(self):
        """
        target: test insert and drop
        method: insert data and drop collection
        expected: verify collection if exist
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        self.create_collection(client, collection_name, default_dim)

        collections = self.list_collections(client)[0]
        assert collection_name in collections

        rng = np.random.default_rng(seed=19530)
        rows = [
            {
                default_primary_key_field_name: i,
                default_vector_field_name: list(rng.random((1, default_dim))[0]),
                default_float_field_name: i * 1.0,
                default_string_field_name: str(i),
            }
            for i in range(ct.default_nb)
        ]
        self.insert(client, collection_name, rows)

        self.drop_collection(client, collection_name)
        collections = self.list_collections(client)[0]
        assert collection_name not in collections

    @pytest.mark.tags(CaseLabel.L2)
    def test_insert_create_index(self):
        """
        target: test insert and create index
        method: 1. insert 2. create index
        expected: verify num entities and index
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        self.create_collection(client, collection_name, default_dim)

        rng = np.random.default_rng(seed=19530)
        rows = [
            {
                default_primary_key_field_name: i,
                default_vector_field_name: list(rng.random((1, default_dim))[0]),
                default_float_field_name: i * 1.0,
                default_string_field_name: str(i),
            }
            for i in range(ct.default_nb)
        ]
        self.insert(client, collection_name, rows)
        self.flush(client, collection_name)

        num_entities = self.get_collection_stats(client, collection_name)[0]
        assert num_entities.get("row_count", None) == ct.default_nb

        # Create index (note: quick setup collection already has index)
        index_params = self.prepare_index_params(client)[0]
        index_params.add_index(default_vector_field_name, index_type="AUTOINDEX")
        self.create_index(client, collection_name, index_params)

        indexes = self.list_indexes(client, collection_name)[0]
        assert default_vector_field_name in indexes
        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L2)
    def test_insert_binary_create_index(self):
        """
        target: test build index insert after vector
        method: insert binary vector and build index
        expected: no error raised
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()

        # 1. Create binary vector collection
        schema = self.create_schema(client, enable_dynamic_field=False)[0]
        schema.add_field(default_primary_key_field_name, DataType.INT64, is_primary=True, auto_id=False)
        schema.add_field(ct.default_binary_vec_field_name, DataType.BINARY_VECTOR, dim=default_dim)
        schema.add_field(default_float_field_name, DataType.FLOAT)
        schema.add_field(default_string_field_name, DataType.VARCHAR, max_length=ct.default_length)

        self.create_collection(client, collection_name, dimension=default_dim, schema=schema)

        # 2. Insert binary data first
        rows = cf.gen_row_data_by_schema(nb=ct.default_nb, schema=schema)
        results = self.insert(client, collection_name, rows)[0]
        assert results["insert_count"] == ct.default_nb

        self.flush(client, collection_name)

        num_entities = self.get_collection_stats(client, collection_name)[0]
        assert num_entities.get("row_count", None) == ct.default_nb

        # 3. Create index after insert
        index_params = self.prepare_index_params(client)[0]
        index_params.add_index(ct.default_binary_vec_field_name, index_type="BIN_IVF_FLAT", metric_type="JACCARD")
        self.create_index(client, collection_name, index_params)

        # 4. Verify index created
        indexes = self.list_indexes(client, collection_name)[0]
        assert ct.default_binary_vec_field_name in indexes

        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L1)
    def test_insert_after_create_index(self):
        """
        target: test insert after create index
        method: 1. create index 2. insert data
        expected: verify index and num entities
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        self.create_collection(client, collection_name, default_dim)

        # Create index first
        index_params = self.prepare_index_params(client)[0]
        index_params.add_index(default_vector_field_name, index_type="AUTOINDEX")
        self.create_index(client, collection_name, index_params)

        indexes = self.list_indexes(client, collection_name)[0]
        assert default_vector_field_name in indexes

        # Then insert data
        rng = np.random.default_rng(seed=19530)
        rows = [
            {
                default_primary_key_field_name: i,
                default_vector_field_name: list(rng.random((1, default_dim))[0]),
                default_float_field_name: i * 1.0,
                default_string_field_name: str(i),
            }
            for i in range(ct.default_nb)
        ]
        self.insert(client, collection_name, rows)

        self.flush(client, collection_name)
        num_entities = self.get_collection_stats(client, collection_name)[0]
        assert num_entities.get("row_count", None) == ct.default_nb
        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L2)
    def test_insert_binary_after_index(self):
        """
        target: test insert binary after index
        method: 1.create index 2.insert binary data
        expected: 1.index ok 2.num entities correct
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()

        # Create binary vector collection
        schema = self.create_schema(client, enable_dynamic_field=True)[0]
        schema.add_field(default_primary_key_field_name, DataType.INT64, is_primary=True, auto_id=False)
        schema.add_field(ct.default_binary_vec_field_name, DataType.BINARY_VECTOR, dim=default_dim)
        schema.add_field(default_float_field_name, DataType.FLOAT)
        schema.add_field(default_string_field_name, DataType.VARCHAR, max_length=ct.default_length)

        index_params = self.prepare_index_params(client)[0]
        index_params.add_index(ct.default_binary_vec_field_name, index_type="BIN_IVF_FLAT", metric_type="HAMMING")

        self.create_collection(client, collection_name, dimension=default_dim, schema=schema, index_params=index_params)

        indexes = self.list_indexes(client, collection_name)[0]
        assert ct.default_binary_vec_field_name in indexes

        # Insert binary data
        rows = cf.gen_row_data_by_schema(nb=ct.default_nb, schema=schema)
        self.insert(client, collection_name, rows)

        self.flush(client, collection_name)
        num_entities = self.get_collection_stats(client, collection_name)[0]
        assert num_entities.get("row_count", None) == ct.default_nb
        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L2)
    def test_insert_auto_id_create_index(self):
        """
        target: test create index in auto_id=True collection
        method: 1.create auto_id=True collection and insert
                2.create index
        expected: index correct
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()

        # Create schema with auto_id=True
        schema = self.create_schema(client, auto_id=True, enable_dynamic_field=True)[0]
        schema.add_field(default_primary_key_field_name, DataType.INT64, is_primary=True, auto_id=True)
        schema.add_field(default_vector_field_name, DataType.FLOAT_VECTOR, dim=default_dim)
        schema.add_field(default_float_field_name, DataType.FLOAT)
        schema.add_field(default_string_field_name, DataType.VARCHAR, max_length=ct.default_length)

        index_params = self.prepare_index_params(client)[0]
        index_params.add_index(default_vector_field_name, index_type="AUTOINDEX")

        self.create_collection(
            client, collection_name, dimension=default_dim, schema=schema, index_params=index_params, auto_id=True
        )

        # Insert without primary key (auto_id)
        rng = np.random.default_rng(seed=19530)
        rows = [
            {
                default_vector_field_name: list(rng.random((1, default_dim))[0]),
                default_float_field_name: i * 1.0,
                default_string_field_name: str(i),
            }
            for i in range(ct.default_nb)
        ]
        results = self.insert(client, collection_name, rows)[0]
        assert results["insert_count"] == ct.default_nb

        self.flush(client, collection_name)
        num_entities = self.get_collection_stats(client, collection_name)[0]
        assert num_entities.get("row_count", None) == ct.default_nb

        indexes = self.list_indexes(client, collection_name)[0]
        assert default_vector_field_name in indexes
        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L2)
    def test_insert_auto_id_true(self, pk_field):
        """
        target: test insert ids fields values when auto_id=True
        method: 1.create collection with auto_id=True 2.insert without ids
        expected: verify primary_keys and num_entities
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()

        # Create schema with auto_id=True and specific primary field
        schema = self.create_schema(client, auto_id=True, enable_dynamic_field=True)[0]
        if pk_field == ct.default_int64_field_name:
            schema.add_field(pk_field, DataType.INT64, is_primary=True, auto_id=True)
        else:
            schema.add_field(pk_field, DataType.VARCHAR, max_length=ct.default_length, is_primary=True, auto_id=True)
        schema.add_field(default_vector_field_name, DataType.FLOAT_VECTOR, dim=default_dim)
        schema.add_field(default_float_field_name, DataType.FLOAT)
        if pk_field != ct.default_string_field_name:
            schema.add_field(default_string_field_name, DataType.VARCHAR, max_length=ct.default_length)

        self.create_collection(client, collection_name, dimension=default_dim, schema=schema, auto_id=True)

        # Insert without primary key (auto_id)
        rng = np.random.default_rng(seed=19530)
        rows = [
            {default_vector_field_name: list(rng.random((1, default_dim))[0]), default_float_field_name: i * 1.0}
            for i in range(ct.default_nb)
        ]
        if pk_field != ct.default_string_field_name:
            for i, row in enumerate(rows):
                row[default_string_field_name] = str(i)

        results = self.insert(client, collection_name, rows)[0]
        assert results["insert_count"] == ct.default_nb
        assert len(set(results["ids"])) == ct.default_nb

        if pk_field == ct.default_int64_field_name:
            assert all(isinstance(i, int) for i in results["ids"])
        else:
            assert all(isinstance(i, str) and i.isdigit() for i in results["ids"])

        self.flush(client, collection_name)
        num_entities = self.get_collection_stats(client, collection_name)[0]
        assert num_entities.get("row_count", None) == ct.default_nb
        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L1)
    def test_insert_twice_auto_id_true(self, pk_field):
        """
        target: test insert ids fields twice when auto_id=True
        method: 1.create collection with auto_id=True 2.insert twice
        expected: verify primary_keys unique
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        nb = 200

        # Create schema with auto_id=True and specific primary field
        schema = self.create_schema(client, auto_id=True, enable_dynamic_field=True)[0]
        if pk_field == ct.default_int64_field_name:
            schema.add_field(pk_field, DataType.INT64, is_primary=True, auto_id=True)
        else:
            schema.add_field(pk_field, DataType.VARCHAR, max_length=ct.default_length, is_primary=True, auto_id=True)
        schema.add_field(default_vector_field_name, DataType.FLOAT_VECTOR, dim=default_dim, nullable=True)
        schema.add_field(default_float_field_name, DataType.FLOAT)
        if pk_field != ct.default_string_field_name:
            schema.add_field(default_string_field_name, DataType.VARCHAR, max_length=ct.default_length)

        self.create_collection(client, collection_name, dimension=default_dim, schema=schema, auto_id=True)

        # Insert twice
        rows = cf.gen_row_data_by_schema(nb, schema=schema, start=0)

        results_1 = self.insert(client, collection_name, rows)[0]
        assert results_1["insert_count"] == nb

        results_2 = self.insert(client, collection_name, rows)[0]
        assert results_2["insert_count"] == nb

        # Verify primary keys are unique across two inserts
        all_ids = list(results_1["ids"]) + list(results_2["ids"])
        assert len(set(all_ids)) == nb * 2

        self.flush(client, collection_name)
        num_entities = self.get_collection_stats(client, collection_name)[0]
        assert num_entities.get("row_count", None) == nb * 2
        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L2)
    def test_insert_auto_id_true_with_provided_pk(self, pk_field):
        """
        target: test insert ids fields values when auto_id=True
        method: 1.create collection with auto_id=True 2.insert with provided pk
        expected: insert failed
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()

        # Create schema with auto_id=True and specific primary field
        schema = self.create_schema(client, auto_id=True, enable_dynamic_field=True)[0]
        if pk_field == ct.default_int64_field_name:
            schema.add_field(pk_field, DataType.INT64, is_primary=True, auto_id=True)
        else:
            schema.add_field(pk_field, DataType.VARCHAR, max_length=ct.default_length, is_primary=True, auto_id=True)
        schema.add_field(default_vector_field_name, DataType.FLOAT_VECTOR, dim=default_dim)
        schema.add_field(default_float_field_name, DataType.FLOAT)
        if pk_field != ct.default_string_field_name:
            schema.add_field(default_string_field_name, DataType.VARCHAR, max_length=ct.default_length)

        self.create_collection(client, collection_name, dimension=default_dim, schema=schema, auto_id=True)

        # Insert with primary key (auto_id)
        rng = np.random.default_rng(seed=19530)
        rows = []
        for i in range(ct.default_nb):
            row = {default_vector_field_name: list(rng.random((1, default_dim))[0]), default_float_field_name: i * 1.0}
            if pk_field == ct.default_int64_field_name:
                row[pk_field] = i
                row[default_string_field_name] = str(i)
            else:
                row[pk_field] = str(i)
            rows.append(row)

        error = {ct.err_code: 1100, ct.err_msg: "more fieldData has pass in: invalid parameter"}
        self.insert(client, collection_name, rows, check_task=CheckTasks.err_res, check_items=error)
        self.drop_collection(client, collection_name)

    @pytest.mark.skip(reason="Covered by test_insert_auto_id_true")
    @pytest.mark.tags(CaseLabel.L2)
    def test_insert_auto_id_true_list_data(self, pk_field):
        """
        target: test insert ids fields values when auto_id=True
        method: 1.create collection with auto_id=True 2.insert list data with ids field values
        expected: assert num entities
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()

        # Create schema with auto_id=True and specific primary field
        schema = self.create_schema(client, auto_id=True, enable_dynamic_field=True)[0]
        if pk_field == ct.default_int64_field_name:
            schema.add_field(pk_field, DataType.INT64, is_primary=True, auto_id=True)
        else:
            schema.add_field(pk_field, DataType.VARCHAR, max_length=ct.default_length, is_primary=True, auto_id=True)
        schema.add_field(default_vector_field_name, DataType.FLOAT_VECTOR, dim=default_dim)
        schema.add_field(default_float_field_name, DataType.FLOAT)
        if pk_field != ct.default_string_field_name:
            schema.add_field(default_string_field_name, DataType.VARCHAR, max_length=ct.default_length)

        self.create_collection(client, collection_name, dimension=default_dim, schema=schema, auto_id=True)

        # Insert without primary key (auto_id)
        rng = np.random.default_rng(seed=19530)
        rows = [
            {default_vector_field_name: list(rng.random((1, default_dim))[0]), default_float_field_name: i * 1.0}
            for i in range(ct.default_nb)
        ]
        if pk_field != ct.default_string_field_name:
            for i, row in enumerate(rows):
                row[default_string_field_name] = str(i)

        results = self.insert(client, collection_name, rows)[0]
        assert results["insert_count"] == ct.default_nb

        self.flush(client, collection_name)
        num_entities = self.get_collection_stats(client, collection_name)[0]
        assert num_entities.get("row_count", None) == ct.default_nb
        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L1)
    def test_insert_with_dataframe_values(self, pk_field, auto_id):
        """
        target: test insert with dataframe data
        method: create collection
        expected: milvus client does not support insert with dataframe
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()

        # Create schema with auto_id
        schema = self.create_schema(client, auto_id=auto_id, enable_dynamic_field=True)[0]
        if pk_field == ct.default_int64_field_name:
            schema.add_field(pk_field, DataType.INT64, is_primary=True, auto_id=auto_id)
        else:
            schema.add_field(pk_field, DataType.VARCHAR, max_length=ct.default_length, is_primary=True, auto_id=auto_id)
        schema.add_field(default_vector_field_name, DataType.FLOAT_VECTOR, dim=default_dim)
        schema.add_field(default_float_field_name, DataType.FLOAT)
        if pk_field != ct.default_string_field_name:
            schema.add_field(default_string_field_name, DataType.VARCHAR, max_length=ct.default_length)

        self.create_collection(client, collection_name, dimension=default_dim, schema=schema, auto_id=auto_id)

        # Try to insert with primary key included (should fail)
        df = cf.gen_default_dataframe_data(nb=100, auto_id=auto_id)
        error = {
            ct.err_code: 999,
            ct.err_msg: f"wrong type of argument 'data',expected 'Dict' or list of 'Dict', got 'DataFrame'",
        }
        self.insert(client, collection_name, df, check_task=CheckTasks.err_res, check_items=error)

        self.flush(client, collection_name)
        num_entities = self.get_collection_stats(client, collection_name)[0]
        assert num_entities.get("row_count", None) == 0
        self.drop_collection(client, collection_name)

    @pytest.mark.skip(reason="Covered by test_insert_auto_id_true")
    @pytest.mark.tags(CaseLabel.L2)
    def test_insert_auto_id_true_with_list_values(self, pk_field):
        """
        target: test insert with auto_id=True
        method: create collection with auto_id=True
        expected: 1.verify num entities 2.verify ids
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        nb = 100

        # Create schema with auto_id=True
        schema = self.create_schema(client, auto_id=True, enable_dynamic_field=True)[0]
        if pk_field == ct.default_int64_field_name:
            schema.add_field(pk_field, DataType.INT64, is_primary=True, auto_id=True)
        else:
            schema.add_field(pk_field, DataType.VARCHAR, max_length=ct.default_length, is_primary=True, auto_id=True)
        schema.add_field(default_vector_field_name, DataType.FLOAT_VECTOR, dim=default_dim)
        schema.add_field(default_float_field_name, DataType.FLOAT)
        if pk_field != ct.default_string_field_name:
            schema.add_field(default_string_field_name, DataType.VARCHAR, max_length=ct.default_length)

        self.create_collection(client, collection_name, dimension=default_dim, schema=schema, auto_id=True)

        # Insert without primary key (auto_id)
        rng = np.random.default_rng(seed=19530)
        rows = [
            {default_vector_field_name: list(rng.random((1, default_dim))[0]), default_float_field_name: i * 1.0}
            for i in range(nb)
        ]
        if pk_field != ct.default_string_field_name:
            for i, row in enumerate(rows):
                row[default_string_field_name] = str(i)

        self.insert(client, collection_name, rows)
        self.flush(client, collection_name)
        num_entities = self.get_collection_stats(client, collection_name)[0]
        assert num_entities.get("row_count", None) == nb
        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L1)
    def test_insert_auto_id_false_same_values(self):
        """
        target: test insert same ids with auto_id false
        method: 1.create collection with auto_id=False 2.insert same int64 field values
        expected: veryfiy insert count
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        nb = 100

        self.create_collection(client, collection_name, default_dim, auto_id=False)

        # Insert with same primary key values
        rng = np.random.default_rng(seed=19530)
        rows = [
            {
                default_primary_key_field_name: 1,
                default_vector_field_name: list(rng.random((1, default_dim))[0]),
                default_float_field_name: i * 1.0,
                default_string_field_name: str(i),
            }
            for i in range(nb)
        ]
        results = self.insert(client, collection_name, rows)[0]
        assert results["insert_count"] == nb
        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L1)
    def test_insert_auto_id_false_negative_values(self):
        """
        target: test insert negative ids with auto_id false
        method: auto_id=False, primary field values is negative
        expected: verify num entities
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        nb = 100

        self.create_collection(client, collection_name, default_dim, auto_id=False)

        # Insert with negative primary key values
        rng = np.random.default_rng(seed=19530)
        rows = [
            {
                default_primary_key_field_name: -i,
                default_vector_field_name: list(rng.random((1, default_dim))[0]),
                default_float_field_name: i * 1.0,
                default_string_field_name: str(i),
            }
            for i in range(nb)
        ]
        results = self.insert(client, collection_name, rows)[0]
        assert results["insert_count"] == nb

        self.flush(client, collection_name)
        num_entities = self.get_collection_stats(client, collection_name)[0]
        assert num_entities.get("row_count", None) == nb
        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L1)
    # @pytest.mark.xfail(reason="issue 15416")
    def test_insert_multi_threading(self):
        """
        target: test concurrent insert
        method: multi threads insert
        expected: verify num entities
        """
        import threading

        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        self.create_collection(client, collection_name, default_dim, consistency_level="Strong")

        thread_num = 4
        threads = []
        rng = np.random.default_rng(seed=19530)
        rows = [
            {
                default_primary_key_field_name: i,
                default_vector_field_name: list(rng.random((1, default_dim))[0]),
                default_float_field_name: i * 1.0,
                default_string_field_name: str(i),
            }
            for i in range(ct.default_nb)
        ]

        def insert(thread_i):
            log.debug(f"In thread-{thread_i}")
            # Adjust primary keys to be unique per thread
            thread_rows = [
                {
                    default_primary_key_field_name: i + thread_i * ct.default_nb,
                    default_vector_field_name: row[default_vector_field_name],
                    default_float_field_name: row[default_float_field_name],
                    default_string_field_name: row[default_string_field_name],
                }
                for i, row in enumerate(rows)
            ]
            results = self.insert(client, collection_name, thread_rows)[0]
            assert results["insert_count"] == ct.default_nb

        for i in range(thread_num):
            x = threading.Thread(target=insert, args=(i,))
            threads.append(x)
            x.start()
        for t in threads:
            t.join()

        self.flush(client, collection_name)
        num_entities = self.get_collection_stats(client, collection_name)[0]
        assert num_entities.get("row_count", None) == ct.default_nb * thread_num
        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L1)
    def test_insert_multi_times(self, dim):
        """
        target: test insert multi times
        method: insert data multi times
        expected: verify num entities
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        step = 120
        nb = 12000

        self.create_collection(client, collection_name, dim, auto_id=False)

        rng = np.random.default_rng(seed=19530)
        start_id = 0
        for _ in range(nb // step):
            rows = [
                {
                    default_primary_key_field_name: i,
                    default_vector_field_name: list(rng.random((1, dim))[0]),
                    default_float_field_name: i * 1.0,
                    default_string_field_name: str(i),
                }
                for i in range(start_id, start_id + step)
            ]
            results = self.insert(client, collection_name, rows)[0]
            assert results["insert_count"] == step
            start_id += step

        self.flush(client, collection_name)
        num_entities = self.get_collection_stats(client, collection_name)[0]
        assert num_entities.get("row_count", None) == nb
        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L1)
    def test_insert_binary_multi_times(self):
        """
        target: test insert entities multi times and final flush
        method: create collection and insert binary entity multi times
        expected: the collection row count equals to nb * nums
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()

        # 1. Create binary vector collection
        schema = self.create_schema(client, enable_dynamic_field=False)[0]
        schema.add_field(default_primary_key_field_name, DataType.INT64, is_primary=True, auto_id=False)
        schema.add_field(ct.default_binary_vec_field_name, DataType.BINARY_VECTOR, dim=default_dim)
        schema.add_field(default_float_field_name, DataType.FLOAT)
        schema.add_field(default_string_field_name, DataType.VARCHAR, max_length=ct.default_length)

        self.create_collection(client, collection_name, dimension=default_dim, schema=schema)

        # 2. Insert binary data multiple times
        nums = 2
        start_id = 0
        for _ in range(nums):
            # Generate data with unique primary keys for each insert
            rows = cf.gen_row_data_by_schema(nb=ct.default_nb, schema=schema, start=start_id)
            results = self.insert(client, collection_name, rows)[0]
            assert results["insert_count"] == ct.default_nb
            start_id += ct.default_nb

        # 3. Verify row count
        self.flush(client, collection_name)
        num_entities = self.get_collection_stats(client, collection_name)[0]
        assert num_entities.get("row_count", None) == ct.default_nb * nums

        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L1)
    def test_insert_all_datatype_collection(self):
        """
        target: test insert into collection that contains all datatype fields
        method: 1.create all datatype collection 2.insert data
        expected: verify num entities
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        nb = 100

        # Create schema with all data types
        schema = cf.gen_all_datatype_collection_schema(dim=default_dim, enable_struct_array_field=False)

        # Create collection
        self.create_collection(client, collection_name, dimension=default_dim, schema=schema)

        # Generate data for all data types
        rows = cf.gen_row_data_by_schema(nb=nb, schema=schema)

        # Insert data
        results = self.insert(client, collection_name, rows)[0]
        assert results["insert_count"] == nb

        # Verify num entities
        self.flush(client, collection_name)
        num_entities = self.get_collection_stats(client, collection_name)[0]
        assert num_entities.get("row_count", None) == nb

        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L2)
    def test_insert_equal_to_resource_limit(self):
        """
        target: test insert data equal to RPC limitation 64MB (67108864)
        method: calculated critical value and insert equivalent data
        expected: insert succeeds
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        # nb = 127583 without json field
        nb = 108993

        self.create_collection(client, collection_name, default_dim, auto_id=False)

        rng = np.random.default_rng(seed=19530)
        rows = [
            {
                default_primary_key_field_name: i,
                default_vector_field_name: list(rng.random((1, default_dim))[0]),
                default_float_field_name: i * 1.0,
                default_string_field_name: str(i),
            }
            for i in range(nb)
        ]
        self.insert(client, collection_name, rows)

        self.flush(client, collection_name)
        num_entities = self.get_collection_stats(client, collection_name)[0]
        assert num_entities.get("row_count", None) == nb
        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L1)
    @pytest.mark.parametrize("nullable", [True, False])
    @pytest.mark.parametrize("default_value_type", ["empty", "none"])
    def test_insert_one_field_using_default_value(self, default_value_type, nullable, auto_id):
        """
        target: test insert with one field using default value
        method: 1. create a collection with one field using default value
                2. insert using default value to replace the field value []/[None]
        expected: insert successfully
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()

        # Create schema with default value field
        schema = self.create_schema(client, auto_id=auto_id, enable_dynamic_field=False)[0]
        if not auto_id:
            schema.add_field(default_primary_key_field_name, DataType.INT64, is_primary=True, auto_id=False)
        else:
            schema.add_field(default_primary_key_field_name, DataType.INT64, is_primary=True, auto_id=True)
        schema.add_field(default_float_field_name, DataType.FLOAT)
        schema.add_field(
            default_string_field_name,
            DataType.VARCHAR,
            max_length=ct.default_length,
            default_value="abc",
            nullable=nullable,
        )
        schema.add_field(default_vector_field_name, DataType.FLOAT_VECTOR, dim=default_dim)

        self.create_collection(client, collection_name, dimension=default_dim, schema=schema, auto_id=auto_id)

        # Insert data with None or omitting the default value field
        rng = np.random.default_rng(seed=19530)
        rows = []
        for i in range(ct.default_nb):
            row = {default_float_field_name: float(i), default_vector_field_name: list(rng.random((1, default_dim))[0])}
            if not auto_id:
                row[default_primary_key_field_name] = i
            if default_value_type == "none":
                row[default_string_field_name] = None
            # If default_value_type == "empty", we don't include the field at all
            rows.append(row)

        self.insert(client, collection_name, rows)
        self.flush(client, collection_name)
        num_entities = self.get_collection_stats(client, collection_name)[0]
        assert num_entities.get("row_count", None) == ct.default_nb
        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L1)
    def test_insert_multi_fields_none_with_default_value(self):
        """
        target: test insert with multi fields include array using none value
        method: 1. create a collection with multi fields using default value
                2. insert using none value to replace the field value
        expected: insert successfully
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        schema = self.create_schema(client)[0]
        dim = 16
        nb = 100
        schema.add_field(default_primary_key_field_name, DataType.INT64, is_primary=True)
        schema.add_field(default_int32_field_name, DataType.INT32, default_value=np.int32(1), nullable=True)
        schema.add_field(default_float_field_name, DataType.FLOAT, default_value=np.float32(1.0), nullable=True)
        schema.add_field(
            default_string_field_name, DataType.VARCHAR, default_value="abc", max_length=100, nullable=True
        )
        schema.add_field(
            "int32_array", datatype=DataType.ARRAY, element_type=DataType.INT32, max_capacity=20, nullable=True
        )
        schema.add_field(
            "float_array", datatype=DataType.ARRAY, element_type=DataType.FLOAT, max_capacity=20, nullable=True
        )
        schema.add_field(
            "string_array",
            datatype=DataType.ARRAY,
            element_type=DataType.VARCHAR,
            max_capacity=20,
            max_length=100,
            nullable=True,
        )
        schema.add_field("json", DataType.JSON, nullable=True)
        schema.add_field(default_float_vec_field_name, DataType.FLOAT_VECTOR, dim=dim)
        self.create_collection(client, collection_name, schema=schema)

        rows = [
            {
                default_primary_key_field_name: i,
                default_int32_field_name: None,
                default_float_field_name: None,
                default_string_field_name: None,
                "int32_array": None,
                "float_array": None,
                "string_array": None,
                "json": None,
                default_float_vec_field_name: cf.gen_vectors(1, dim=dim)[0],
            }
            for i in range(nb)
        ]
        self.insert(client, collection_name, rows)
        self.flush(client, collection_name)
        num_entities = self.get_collection_stats(client, collection_name)[0]
        assert num_entities.get("row_count", None) == nb

        # build index and load
        index_params = self.prepare_index_params(client)[0]
        index_params.add_index(default_float_vec_field_name, metric_type="L2")
        self.create_index(client, collection_name, index_params)
        self.load_collection(client, collection_name)

        # try to query None value entities, should be empty
        res, _ = self.query(client, collection_name, filter=f"{default_string_field_name} is null")
        assert len(res) == 0

        # try to query default value entities, should be not empty
        res, _ = self.query(client, collection_name, filter=f"{default_string_field_name}=='abc'")
        assert len(res) == nb

        # try to query None value entities on json field, should not be empty
        res, _ = self.query(client, collection_name, filter=f"json is null")
        assert len(res) == nb

        res, _ = self.query(client, collection_name, filter=f"int32_array is null")
        assert len(res) == nb

        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L1)
    @pytest.mark.parametrize("sparse_format", ["csr_matrix", "csr_array"])
    def test_milvus_client_insert_sparse_vector_scipy(self, sparse_format):
        """
        target: test insert and search sparse vectors using scipy.sparse csr format directly
        method: insert sparse vectors as scipy.sparse csr matrices per row, then search with csr query
        expected: insert and search succeed with correct results
        """
        from scipy import sparse as sp

        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        nb = 1000
        dim = 10000

        # 1. Create schema with sparse vector field
        schema = self.create_schema(client, enable_dynamic_field=False)[0]
        schema.add_field(ct.default_int64_field_name, DataType.INT64, is_primary=True, auto_id=False)
        schema.add_field(ct.default_sparse_vec_field_name, DataType.SPARSE_FLOAT_VECTOR)
        index_params = self.prepare_index_params(client)[0]
        index_params.add_index(ct.default_sparse_vec_field_name, index_type="SPARSE_INVERTED_INDEX", metric_type="IP")
        self.create_collection(client, collection_name, schema=schema, index_params=index_params)

        # 2. Build scipy.sparse constructor
        sparse_cls = getattr(sp, sparse_format)

        # 3. Generate sparse data as scipy.sparse csr single-row matrices per row
        rng = np.random.default_rng(seed=19530)
        rows = []
        for i in range(nb):
            nnz = rng.integers(20, 30)
            indices = sorted(rng.choice(dim, size=nnz, replace=False))
            values = rng.random(nnz).astype(np.float32)
            row_sparse = sparse_cls((values, indices, [0, nnz]), shape=(1, dim))
            rows.append({ct.default_int64_field_name: i, ct.default_sparse_vec_field_name: row_sparse})
        self.insert(client, collection_name, rows)
        self.flush(client, collection_name)
        self.load_collection(client, collection_name)

        # 4. Search with scipy.sparse query vector
        q_nnz = 25
        q_indices = sorted(rng.choice(dim, size=q_nnz, replace=False))
        q_values = rng.random(q_nnz).astype(np.float32)
        query_sparse = sparse_cls((q_values, q_indices, [0, q_nnz]), shape=(1, dim))
        self.search(
            client,
            collection_name,
            data=[query_sparse],
            anns_field=ct.default_sparse_vec_field_name,
            limit=default_limit,
            search_params={"metric_type": "IP"},
            check_task=CheckTasks.check_search_results,
            check_items={
                "enable_milvus_client_api": True,
                "nq": 1,
                "limit": default_limit,
                "pk_name": ct.default_int64_field_name,
            },
        )

        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L2)
    @pytest.mark.parametrize("sparse_format", ["csc_matrix", "coo_matrix", "dok_matrix", "lil_matrix", "coo_array"])
    def test_milvus_client_insert_sparse_vector_scipy_to_csr(self, sparse_format):
        """
        target: test insert sparse vectors created in non-csr scipy.sparse formats via .tocsr() conversion
        method: create sparse data in various scipy formats, convert to csr, insert and search
        expected: insert and search succeed after converting to csr
        """
        from scipy import sparse as sp

        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        nb = 1000
        dim = 10000

        # 1. Create schema with sparse vector field
        schema = self.create_schema(client, enable_dynamic_field=False)[0]
        schema.add_field(ct.default_int64_field_name, DataType.INT64, is_primary=True, auto_id=False)
        schema.add_field(ct.default_sparse_vec_field_name, DataType.SPARSE_FLOAT_VECTOR)
        index_params = self.prepare_index_params(client)[0]
        index_params.add_index(ct.default_sparse_vec_field_name, index_type="SPARSE_INVERTED_INDEX", metric_type="IP")
        self.create_collection(client, collection_name, schema=schema, index_params=index_params)

        # 2. Build scipy.sparse constructor for the non-csr format
        sparse_cls = getattr(sp, sparse_format)

        # 3. Generate sparse data: create in target format, then .tocsr() for insert
        #    Non-csr formats lack .indices/.data attributes required by pymilvus per-row path
        rng = np.random.default_rng(seed=19530)
        rows = []
        for i in range(nb):
            nnz = rng.integers(20, 30)
            indices = sorted(rng.choice(dim, size=nnz, replace=False))
            values = rng.random(nnz).astype(np.float32)
            row_sparse = sparse_cls(sp.csr_matrix((values, indices, [0, nnz]), shape=(1, dim))).tocsr()
            rows.append({ct.default_int64_field_name: i, ct.default_sparse_vec_field_name: row_sparse})
        self.insert(client, collection_name, rows)
        self.flush(client, collection_name)
        self.load_collection(client, collection_name)

        # 4. Search with scipy.sparse csr query vector
        q_nnz = 25
        q_indices = sorted(rng.choice(dim, size=q_nnz, replace=False))
        q_values = rng.random(q_nnz).astype(np.float32)
        query_sparse = sp.csr_matrix((q_values, q_indices, [0, q_nnz]), shape=(1, dim))
        self.search(
            client,
            collection_name,
            data=[query_sparse],
            anns_field=ct.default_sparse_vec_field_name,
            limit=default_limit,
            search_params={"metric_type": "IP"},
            check_task=CheckTasks.check_search_results,
            check_items={
                "enable_milvus_client_api": True,
                "nq": 1,
                "limit": default_limit,
                "pk_name": ct.default_int64_field_name,
            },
        )

        self.drop_collection(client, collection_name)


class TestMilvusClientInsertString(TestMilvusClientV2Base):
    """
    ******************************************************************
    The following cases are used to test insert string
    ******************************************************************
    """

    @pytest.mark.tags(CaseLabel.L0)
    def test_milvus_client_insert_string_field_is_primary(self):
        """
        target: test insert string is primary
        method: 1.create a collection and string field is primary
                2.insert string field data
        expected: Insert Successfully
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()

        # 1. Create schema with string field as primary key
        schema = self.create_schema(client, enable_dynamic_field=False)[0]
        schema.add_field(ct.default_int64_field_name, DataType.INT64)
        schema.add_field(default_float_field_name, DataType.FLOAT)
        schema.add_field(
            ct.default_string_field_name, DataType.VARCHAR, max_length=ct.default_length, is_primary=True, auto_id=False
        )
        schema.add_field(default_vector_field_name, DataType.FLOAT_VECTOR, dim=default_dim)

        # 2. Create collection
        self.create_collection(client, collection_name, dimension=default_dim, schema=schema)

        # 3. Generate row data
        rows = cf.gen_row_data_by_schema(nb=ct.default_nb, schema=schema)

        # 4. Extract primary keys (string field values) for verification
        expected_primary_keys = [row[ct.default_string_field_name] for row in rows]

        # 5. Insert data
        results = self.insert(client, collection_name, rows)[0]
        assert results["insert_count"] == ct.default_nb
        assert results["ids"] == expected_primary_keys

        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L0)
    @pytest.mark.parametrize(
        "string_fields",
        [
            [
                cf.gen_string_field(name="string_field1"),
                cf.gen_string_field(name="string_field2"),
                cf.gen_string_field(name="string_field3"),
            ]
        ],
    )
    def test_milvus_client_insert_multi_string_fields(self, string_fields):
        """
        target: test insert multi string fields
        method: 1.create a collection
                2.Insert multi string fields
        expected: Insert Successfully
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()

        # 1. Create schema with multiple string fields
        schema = self.create_schema(client, enable_dynamic_field=False)[0]
        schema.add_field(ct.default_int64_field_name, DataType.INT64, is_primary=True, auto_id=False)
        schema.add_field(default_float_field_name, DataType.FLOAT)
        schema.add_field(ct.default_string_field_name, DataType.VARCHAR, max_length=ct.default_length)
        schema.add_field(default_vector_field_name, DataType.FLOAT_VECTOR, dim=default_dim)
        # Add additional string fields
        for field in string_fields:
            schema.add_field(field.name, DataType.VARCHAR, max_length=ct.default_length)

        # 2. Create collection
        self.create_collection(client, collection_name, dimension=default_dim, schema=schema)

        # 3. Generate row data
        rows = cf.gen_row_data_by_schema(nb=ct.default_nb, schema=schema)

        # 4. Add random string values for additional string fields (matching gen_dataframe_multi_string_fields behavior)
        for field in string_fields:
            if field.dtype == DataType.VARCHAR:
                string_values = cf.gen_string(default_nb)
                for i, row in enumerate(rows):
                    row[field.name] = string_values[i]

        # 5. Insert data
        results = self.insert(client, collection_name, rows)[0]
        assert results["insert_count"] == ct.default_nb

        # 6. Verify num entities
        self.flush(client, collection_name)
        num_entities = self.get_collection_stats(client, collection_name)[0]
        assert num_entities.get("row_count", None) == ct.default_nb

        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L0)
    def test_milvus_client_insert_string_field_length_exceed(self):
        """
        target: test insert string field exceed the maximum length
        method: 1.create a collection with VARCHAR field max_length=65535
                2.Insert string field length is exceeded maximum value of 65535
        expected: Raise exceptions
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()

        # 1. Create collection with explicit schema (VARCHAR field max_length=65535)
        schema = self.create_schema(client, enable_dynamic_field=False)[0]
        schema.add_field(default_primary_key_field_name, DataType.INT64, is_primary=True, auto_id=False)
        schema.add_field(default_vector_field_name, DataType.FLOAT_VECTOR, dim=default_dim)
        schema.add_field(default_float_field_name, DataType.FLOAT)
        schema.add_field(default_string_field_name, DataType.VARCHAR, max_length=65535)
        self.create_collection(client, collection_name, dimension=default_dim, schema=schema)

        # 2. Generate row data with string field exceeding max length (65535)
        max_length = 65535
        rng = np.random.default_rng(seed=19530)
        # Generate a string that exceeds max length
        long_string = cf.gen_str_by_length(length=max_length + 1)

        # Generate normal data for one row
        rows = [
            {
                default_primary_key_field_name: 0,
                default_vector_field_name: list(rng.random((1, default_dim))[0]),
                default_float_field_name: 0.0,
                default_string_field_name: long_string,
            }
        ]

        # 3. Verify error on insert
        error = {ct.err_code: 1100, ct.err_msg: "length of varchar field varchar exceeds max length"}
        self.insert(client, collection_name, data=rows, check_task=CheckTasks.err_res, check_items=error)

        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L1)
    @pytest.mark.parametrize("str_field_value", ["", "    "])
    def test_milvus_client_insert_string_field_space_empty(self, str_field_value):
        """
        target: test create collection with string field
        method: 1.create a collection
                2.Insert string field with space or empty string
        expected: Insert successfully
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        nb = 100

        # 1. Create collection with default schema
        self.create_collection(client, collection_name, default_dim)

        # 2. Generate row data with string field set to empty or space-only string
        rng = np.random.default_rng(seed=19530)
        rows = [
            {
                default_primary_key_field_name: i,
                default_vector_field_name: list(rng.random((1, default_dim))[0]),
                default_float_field_name: i * 1.0,
                default_string_field_name: str_field_value,  # Empty string or space-only string
            }
            for i in range(nb)
        ]

        # 3. Insert data
        results = self.insert(client, collection_name, rows)[0]
        assert results["insert_count"] == nb

        # 4. Verify num entities
        self.flush(client, collection_name)
        num_entities = self.get_collection_stats(client, collection_name)[0]
        assert num_entities.get("row_count", None) == nb

        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L1)
    @pytest.mark.parametrize("str_field_value", ["", "    "])
    def test_milvus_client_insert_string_field_is_pk_and_empty(self, str_field_value):
        """
        target: test create collection with string field is primary
        method: 1.create a collection
                2.Insert string field with empty or space-only string, string field is pk
        expected: Insert successfully
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        nb = 100

        # 1. Create schema with string field as primary key (matching gen_string_pk_default_collection_schema)
        schema = self.create_schema(client, enable_dynamic_field=False)[0]
        schema.add_field(ct.default_int64_field_name, DataType.INT64)
        schema.add_field(default_float_field_name, DataType.FLOAT)
        schema.add_field(
            ct.default_string_field_name, DataType.VARCHAR, max_length=ct.default_length, is_primary=True, auto_id=False
        )
        schema.add_field(default_vector_field_name, DataType.FLOAT_VECTOR, dim=default_dim)

        # 2. Create collection
        self.create_collection(client, collection_name, dimension=default_dim, schema=schema)

        # 3. Generate row data with string field (primary key) set to empty or space-only string
        rng = np.random.default_rng(seed=19530)
        rows = [
            {
                ct.default_int64_field_name: i,
                default_float_field_name: i * 1.0,
                ct.default_string_field_name: str_field_value,  # Empty string or space-only string as primary key
                default_vector_field_name: list(rng.random((1, default_dim))[0]),
            }
            for i in range(nb)
        ]

        # 4. Insert data
        results = self.insert(client, collection_name, rows)[0]
        assert results["insert_count"] == nb

        # 5. Verify num entities
        self.flush(client, collection_name)
        num_entities = self.get_collection_stats(client, collection_name)[0]
        assert num_entities.get("row_count", None) == nb

        self.drop_collection(client, collection_name)


class TestMilvusClientInsertArray(TestMilvusClientV2Base):
    """
    ******************************************************************
    The following cases are used to test insert array
    ******************************************************************
    """

    def gen_array_collection_schema(
        self,
        description=ct.default_desc,
        primary_field=ct.default_int64_field_name,
        auto_id=False,
        dim=ct.default_dim,
        enable_dynamic_field=False,
        max_capacity=ct.default_max_capacity,
        max_length=100,
        with_json=False,
        **kwargs,
    ):
        """
        Generate array collection schema.
        """
        schema = MilvusClient.create_schema(
            auto_id=auto_id, enable_dynamic_field=enable_dynamic_field, description=description, **kwargs
        )

        # Add primary key field
        if primary_field == ct.default_int64_field_name:
            schema.add_field(
                field_name=ct.default_int64_field_name, datatype=DataType.INT64, is_primary=True, auto_id=auto_id
            )
        elif primary_field == ct.default_string_field_name:
            schema.add_field(
                field_name=ct.default_string_field_name,
                datatype=DataType.VARCHAR,
                max_length=ct.default_length,
                is_primary=True,
                auto_id=auto_id,
            )
        else:
            log.error("Primary key only support int or varchar")
            assert False

        # Add vector field
        schema.add_field(field_name=ct.default_float_vec_field_name, datatype=DataType.FLOAT_VECTOR, dim=dim)

        if not enable_dynamic_field:
            # Add JSON field if requested
            if with_json:
                schema.add_field(field_name=ct.default_json_field_name, datatype=DataType.JSON, nullable=True)

            # Add array fields
            schema.add_field(
                field_name=ct.default_int32_array_field_name,
                datatype=DataType.ARRAY,
                element_type=DataType.INT32,
                max_capacity=max_capacity,
            )
            schema.add_field(
                field_name=ct.default_float_array_field_name,
                datatype=DataType.ARRAY,
                element_type=DataType.FLOAT,
                max_capacity=max_capacity,
            )
            schema.add_field(
                field_name=ct.default_string_array_field_name,
                datatype=DataType.ARRAY,
                element_type=DataType.VARCHAR,
                max_capacity=max_capacity,
                max_length=max_length,
                nullable=True,
            )

        return schema

    @pytest.mark.tags(CaseLabel.L1)
    @pytest.mark.parametrize("auto_id", [True, False])
    def test_milvus_client_insert_array_data(self, auto_id):
        """
        target: test insert data with array fields
        method: Insert data with array fields
        expected: assert num entities
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()

        # 1. Create schema
        schema = self.gen_array_collection_schema(auto_id=auto_id)

        # 2. Create collection
        self.create_collection(client, collection_name, schema=schema)

        # 3. Generate row data with array fields
        rows = cf.gen_row_data_by_schema(nb=default_nb, schema=schema)

        # 4. Insert data
        results = self.insert(client, collection_name, rows)[0]
        assert results["insert_count"] == ct.default_nb

        # 5. Verify num entities
        self.flush(client, collection_name)
        num_entities = self.get_collection_stats(client, collection_name)[0]
        assert num_entities.get("row_count", None) == ct.default_nb

        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L2)
    def test_milvus_client_insert_array_empty_field(self):
        """
        target: test insert data with empty array field
        method: 1.create collection with array fields
                2.insert data with int32_array field set to empty list []
        expected: insert successfully and verify num entities
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        nb = ct.default_nb

        # 1. Create schema
        schema = self.gen_array_collection_schema()

        # 2. Create collection
        self.create_collection(client, collection_name, schema=schema)

        # 3. Generate row data with array fields, set int32_array to empty lists
        rows = cf.gen_row_data_by_schema(nb=nb, schema=schema)
        # Set int32_array field to empty lists
        for row in rows:
            row[ct.default_int32_array_field_name] = []

        # 4. Insert data
        results = self.insert(client, collection_name, rows)[0]
        assert results["insert_count"] == nb

        # 5. Verify num entities
        self.flush(client, collection_name)
        num_entities = self.get_collection_stats(client, collection_name)[0]
        assert num_entities.get("row_count", None) == nb

        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L2)
    def test_milvus_client_insert_array_length_differ(self):
        """
        target: test insert row data with different array lengths
        method: 1.create collection with array fields
                2.insert data with every row's array length differ
        expected: insert successfully and verify num entities
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        nb = ct.default_nb
        array_length = ct.default_max_capacity

        # 1. Create schema
        schema = self.gen_array_collection_schema(max_capacity=array_length)

        # 2. Create collection
        self.create_collection(client, collection_name, schema=schema)

        # 3. Generate row data with different array lengths for each row
        rows = []
        for i in range(nb):
            arr_len1 = random.randint(0, array_length)
            arr_len2 = random.randint(0, array_length)
            row = {
                ct.default_int64_field_name: i,
                ct.default_float_vec_field_name: [random.random() for _ in range(default_dim)],
                ct.default_int32_array_field_name: [np.int32(j) for j in range(arr_len1)],
                ct.default_float_array_field_name: [np.float32(j) for j in range(arr_len2)],
                ct.default_string_array_field_name: [str(j) for j in range(array_length)],
            }
            rows.append(row)

        # 4. Insert data
        results = self.insert(client, collection_name, rows)[0]
        assert results["insert_count"] == nb

        # 5. Verify num entities
        self.flush(client, collection_name)
        num_entities = self.get_collection_stats(client, collection_name)[0]
        assert num_entities.get("row_count", None) == nb

        # 6. Upsert 2 rows (matching original test)
        upsert_data = cf.gen_row_data_by_schema(nb=2, schema=schema)
        self.upsert(client, collection_name, upsert_data)

        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L2)
    def test_milvus_client_insert_array_length_invalid(self):
        """
        target: test insert array with length exceeding max_capacity
        method: 1.create collection with array fields
                2.insert data with array length > max_capacity
        expected: raise error
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        nb = 11
        dim = 32
        array_length = ct.default_max_capacity

        # 1. Create schema
        schema = self.gen_array_collection_schema(dim=dim, max_capacity=array_length)

        # 2. Create collection
        self.create_collection(client, collection_name, schema=schema)

        # 3. Generate row data
        rows = cf.gen_row_data_by_schema(nb=nb, schema=schema)

        # 4. Set array length > max_capacity for the 2nd row (index 1)
        arr_len = array_length + 1
        rows[1][ct.default_float_array_field_name] = [np.float32(i) for i in range(arr_len)]

        # 5. Verify error on insert
        err_msg = f"the length ({arr_len}) of 1th array exceeds max capacity ({array_length})"
        error = {ct.err_code: 1100, ct.err_msg: err_msg}
        self.insert(client, collection_name, data=rows, check_task=CheckTasks.err_res, check_items=error)

        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L2)
    def test_milvus_client_insert_array_type_invalid(self):
        """
        target: test insert array with invalid element type
        method: 1.insert string values to an int array
                2.upsert float values to a string array
        expected: raise error
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        arr_len = 5
        nb = 10
        dim = 8

        # 1. Create schema
        schema = self.gen_array_collection_schema(dim=dim)

        # 2. Create collection
        self.create_collection(client, collection_name, schema=schema)

        # 3. Test 1: Insert string values to an int array
        rows = cf.gen_row_data_by_schema(nb=nb, schema=schema)
        rows[1][ct.default_int32_array_field_name] = [str(i) for i in range(arr_len)]
        err_msg = "The Input data type is inconsistent with defined schema"
        error = {ct.err_code: 999, ct.err_msg: err_msg}
        self.insert(client, collection_name, data=rows, check_task=CheckTasks.err_res, check_items=error)

        # 4. Test 2: Upsert float values to a string array
        rows = cf.gen_row_data_by_schema(nb=nb, schema=schema)
        rows[1][ct.default_string_array_field_name] = [np.float32(i) for i in range(arr_len)]
        self.upsert(client, collection_name, data=rows, check_task=CheckTasks.err_res, check_items=error)

        self.drop_collection(client, collection_name)

    @pytest.mark.tags(CaseLabel.L2)
    def test_milvus_client_insert_array_mixed_value(self):
        """
        target: test insert array consisting of mixed values
        method: insert array consisting of mixed values (string, int, list, bool)
        expected: raise error
        """
        client = self._client()
        collection_name = cf.gen_collection_name_by_testcase_name()
        nb = 10
        dim = 32

        # 1. Create schema
        schema = self.gen_array_collection_schema(dim=dim)

        # 2. Create collection
        self.create_collection(client, collection_name, schema=schema)

        # 3. Generate row data
        rows = cf.gen_row_data_by_schema(nb=nb, schema=schema)

        # 4. Set array consisting of mixed values (string, int, list, bool) for the 2nd row (index 1)
        rows[1][ct.default_string_array_field_name] = ["a", 1, [2.0, 3.0], False]

        # 5. Verify error on insert
        err_msg = "The Input data type is inconsistent with defined schema"
        error = {ct.err_code: 999, ct.err_msg: err_msg}
        self.insert(client, collection_name, data=rows, check_task=CheckTasks.err_res, check_items=error)

        self.drop_collection(client, collection_name)