test: add more test case on partial update (#46628)

Issue: #46627  
add one more test case to cover duplicate pk partial update

 On branch feature/partial-update
 Changes to be committed:
	modified:   milvus_client/test_milvus_client_partial_update.py

<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
- Core invariant: upserts with partial_update=True consolidate records
by primary key (PK) rather than creating duplicate rows; this test
verifies the partial-update upsert path preserves PK identity and merge
semantics.
- Change: adds test
test_milvus_client_partial_update_duplicate_pk_partial_update which
inserts duplicate-PK batches, then calls client.upsert(...,
partial_update=True) on a subset of fields and asserts final row count
equals default_nb, exercising the partial-update code path (upsert →
partial update handling → query) not previously covered.
- No production logic removed/simplified: this PR only adds test
coverage (no code paths removed or altered); nothing in production code
is changed or simplified by the PR.
- No data loss or regression introduced: the test validates concrete
code paths — upsert with partial_update True followed by
query(out_fields/with_vec, pk checks) — and asserts deduplication
(2×default_nb → default_nb). Because the PR only adds assertions against
existing behavior and does not modify runtime logic, it cannot cause
data loss or behavioral regressions.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Signed-off-by: Eric Hou <eric.hou@zilliz.com>
Co-authored-by: Eric Hou <eric.hou@zilliz.com>

tests/python_client/milvus_client/test_milvus_client_partial_update.py

@@ -1321,6 +1321,50 @@ class TestMilvusClientPartialUpdateValid(TestMilvusClientV2Base):
 
         self.drop_collection(client, collection_name)
 
+    @pytest.mark.tags(CaseLabel.L1)
+    def test_milvus_client_partial_update_duplicate_pk_partial_update(self):
+        """
+        target: test PU will success when partial update duplicate pk with partial update
+        method:
+            1. Create a collection
+            2. Insert rows with duplicate pk
+            3. Upsert the rows with duplicate pk with partial update
+        expected: Step 3 result in default_nb rows
+        """
+        # step 1: create collection
+        client = self._client()
+        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_int32_field_name, DataType.INT32, nullable=True)
+        index_params = self.prepare_index_params(client)[0]
+        index_params.add_index(default_primary_key_field_name, index_type="AUTOINDEX")
+        index_params.add_index(default_vector_field_name, index_type="AUTOINDEX")
+        index_params.add_index(default_int32_field_name, index_type="AUTOINDEX")
+        collection_name = cf.gen_collection_name_by_testcase_name(module_index=1)
+        self.create_collection(client, collection_name, default_dim, schema=schema, 
+                               consistency_level="Strong", index_params=index_params)
+        
+        # step 2: Insert rows with duplicate pk
+        rows = cf.gen_row_data_by_schema(nb=default_nb, schema=schema, skip_field_names=[default_int32_field_name])
+        self.insert(client, collection_name, rows)
+        result = self.query(client, collection_name, filter=default_search_exp, output_fields=["count(*)"])
+        assert result[0][0]["count(*)"] == default_nb
+
+        dup_rows = cf.gen_row_data_by_schema(nb=default_nb, schema=schema, skip_field_names=[default_int32_field_name])
+        self.insert(client, collection_name, dup_rows)
+        result = self.query(client, collection_name, filter=default_search_exp, output_fields=["count(*)"])
+        assert result[0][0]["count(*)"] == default_nb * 2
+        
+        # step 3: Upsert the rows with duplicate pk with partial update
+        new_rows = cf.gen_row_data_by_schema(nb=default_nb, schema=schema, 
+                                            desired_field_names=[default_primary_key_field_name, default_int32_field_name])
+        self.upsert(client, collection_name, new_rows, partial_update=True)
+        result = self.query(client, collection_name, filter=default_search_exp, output_fields=["count(*)"])
+        assert result[0][0]["count(*)"] == default_nb
+        
+        self.drop_collection(client, collection_name)
+        
 class TestMilvusClientPartialUpdateInvalid(TestMilvusClientV2Base):
     """ Test case of partial update interface """
     @pytest.fixture(scope="function", params=[False, True])