enhance: general performance improvements across MilvusClient path (#3279)

## Summary

- **Fix duplicate `@retry_on_rpc_failure`** on `search()` in
`grpc_handler.py` — was causing up to 75×75=5625 retry attempts instead
of 75
- **24 micro-optimizations** across the MilvusClient/AsyncMilvusClient
code path (no refactoring, no API changes):
- Replace O(n²) bytes concatenation with `b"".join` and `struct.pack`
batch (~136x faster at n=1000)
  - Use `orjson.loads` instead of stdlib `json.loads` (~6.7x faster)
- Replace `deepcopy` with shallow `dict()` for search params (~4.2x
faster)
- Use `frozenset` for O(1) membership checks instead of list O(n) (~3.6x
faster)
  - Use `WhichOneof` instead of cascading `HasField` in `len_of`
- Use `time.monotonic` + lazy `traceback.format_exc` in error_handler
(~13-17x faster)
- Use builtin `isinstance(x, dict/list)` instead of `typing.Dict/List`
(~3.3-6.5x faster)
- Cache protobuf attribute chains to avoid repeated traversals (~1.3x
faster)
  - Use dict dispatch tables instead of if/elif chains
- Move per-call dict/function creation to module-level constants (~4.5x
faster)
- Use `template.copy()` instead of `dict.fromkeys` per iteration (~5.4x
faster)
- Skip dynamic field json comprehension when dynamic is disabled (~16x
faster)
  - Consolidate `extend([item])` to `append` (~2.8x faster)
- **80 benchmark tests** added in
`tests/benchmark/test_perf_improvements.py` covering all optimization
patterns
- Net **-536 / +1562 lines** across 11 files (10 production + 1
benchmark test file)
- All 2126 existing unit tests pass, all 80 benchmark tests pass

## Test plan

- [x] `make lint` passes (black + ruff clean)
- [x] `make unittest` — 2126 passed, 4 skipped, 1 xfailed
- [x] `pytest tests/benchmark/test_perf_improvements.py` — 80 benchmarks
pass
- [ ] CI checks pass

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Signed-off-by: yangxuan <xuan.yang@zilliz.com>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: XuanYang-cn

pymilvus/client/async_grpc_handler.py

@@ -1,13 +1,13 @@
 import asyncio
 import base64
-import json
 import socket
 import time
 from pathlib import Path
 from typing import Dict, List, Optional, Tuple, Union
 from urllib import parse
 
 import grpc
+import orjson
 from grpc._cython import cygrpc
 
 from pymilvus.client.call_context import CallContext, _api_level_md
@@ -1414,7 +1414,8 @@ async def query(
         _, dynamic_fields = entity_helper.extract_dynamic_field_from_result(response)
         keys = [field_data.field_name for field_data in response.fields_data]
         filtered_keys = [k for k in keys if k != "$meta"]
-        results = [dict.fromkeys(filtered_keys) for _ in range(num_entities)]
+        template = dict.fromkeys(filtered_keys)
+        results = [template.copy() for _ in range(num_entities)]
         lazy_field_data = []
         for field_data in response.fields_data:
             lazy_extracted = entity_helper.extract_row_data_from_fields_data_v2(field_data, results)
@@ -1620,7 +1621,7 @@ async def describe_index(
             info_dict["field_name"] = response.index_descriptions[0].field_name
             info_dict["index_name"] = response.index_descriptions[0].index_name
             if info_dict.get("params"):
-                info_dict["params"] = json.loads(info_dict["params"])
+                info_dict["params"] = orjson.loads(info_dict["params"])
             info_dict["total_rows"] = response.index_descriptions[0].total_rows
             info_dict["indexed_rows"] = response.index_descriptions[0].indexed_rows
             info_dict["pending_index_rows"] = response.index_descriptions[0].pending_index_rows

pymilvus/client/entity_helper.py

@@ -1,5 +1,4 @@
 import base64
-import json
 import logging
 import socket
 import threading
@@ -9,6 +8,7 @@
 from urllib import parse
 
 import grpc
+import orjson
 from grpc._cython import cygrpc
 
 from pymilvus.client.call_context import CallContext, _api_level_md
@@ -1221,7 +1221,6 @@ def _execute_hybrid_search(
                 return SearchFuture(None, None, e)
             raise e from e
 
-    @retry_on_rpc_failure()
     @retry_on_rpc_failure()
     def search(
         self,
@@ -1590,7 +1589,7 @@ def describe_index(
             info_dict["field_name"] = response.index_descriptions[0].field_name
             info_dict["index_name"] = response.index_descriptions[0].index_name
             if info_dict.get("params"):
-                info_dict["params"] = json.loads(info_dict["params"])
+                info_dict["params"] = orjson.loads(info_dict["params"])
             info_dict["total_rows"] = response.index_descriptions[0].total_rows
             info_dict["indexed_rows"] = response.index_descriptions[0].indexed_rows
             info_dict["pending_index_rows"] = response.index_descriptions[0].pending_index_rows
@@ -2201,7 +2200,8 @@ def query(
 
         keys = [field_data.field_name for field_data in response.fields_data]
         filtered_keys = [k for k in keys if k != "$meta"]
-        results = [dict.fromkeys(filtered_keys) for _ in range(num_entities)]
+        template = dict.fromkeys(filtered_keys)
+        results = [template.copy() for _ in range(num_entities)]
         lazy_field_data = []
         for field_data in response.fields_data:
             lazy_extracted = entity_helper.extract_row_data_from_fields_data_v2(field_data, results)

pymilvus/client/grpc_handler.py

@@ -57,6 +57,16 @@
 )
 from .utils import get_params, traverse_info, traverse_upsert_info
 
+_JSON_TYPE_MAP = {
+    DataType.INT8: "Int8",
+    DataType.INT16: "Int16",
+    DataType.INT32: "Int32",
+    DataType.INT64: "Int64",
+    DataType.BOOL: "Bool",
+    DataType.VARCHAR: "VarChar",
+    DataType.STRING: "VarChar",
+}
+
 
 class Prepare:
     @classmethod
@@ -558,7 +568,7 @@ def _function_output_field_names(fields_info: List[Dict]):
 
     @staticmethod
     def _num_input_fields(fields_info: List[Dict], is_upsert: bool):
-        return len([field for field in fields_info if Prepare._is_input_field(field, is_upsert)])
+        return sum(1 for field in fields_info if Prepare._is_input_field(field, is_upsert))
 
     @staticmethod
     def _process_struct_field(
@@ -784,7 +794,7 @@ def _parse_row_request(
 
         try:
             for entity in entities:
-                if not isinstance(entity, Dict):
+                if not isinstance(entity, dict):
                     msg = f"expected Dict, got '{type(entity).__name__}'"
                     raise TypeError(msg)
                 for k, v in entity.items():
@@ -838,13 +848,12 @@ def _parse_row_request(
                         raise DataNotMatchException(
                             message=ExceptionsMessage.InsertMissedField % key
                         )
-                json_dict = {
-                    k: v
-                    for k, v in entity.items()
-                    if k not in fields_data and k not in struct_fields_data and enable_dynamic
-                }
-
                 if enable_dynamic:
+                    json_dict = {
+                        k: v
+                        for k, v in entity.items()
+                        if k not in fields_data and k not in struct_fields_data
+                    }
                     json_value = entity_helper.convert_to_json(json_dict)
                     d_field.scalars.json_data.data.append(json_value)
 
@@ -939,7 +948,7 @@ def _parse_upsert_row_request(
 
         try:
             for entity in entities:
-                if not isinstance(entity, Dict):
+                if not isinstance(entity, dict):
                     msg = f"expected Dict, got '{type(entity).__name__}'"
                     raise TypeError(msg)
                 for k, v in entity.items():
@@ -999,13 +1008,12 @@ def _parse_upsert_row_request(
                         raise DataNotMatchException(
                             message=ExceptionsMessage.InsertMissedField % key
                         )
-                json_dict = {
-                    k: v
-                    for k, v in entity.items()
-                    if k not in fields_data and k not in struct_fields_data and enable_dynamic
-                }
-
                 if enable_dynamic:
+                    json_dict = {
+                        k: v
+                        for k, v in entity.items()
+                        if k not in fields_data and k not in struct_fields_data
+                    }
                     json_value = entity_helper.convert_to_json(json_dict)
                     d_field.scalars.json_data.data.append(json_value)
                     field_len[DYNAMIC_FIELD_NAME] += 1
@@ -1049,7 +1057,7 @@ def _parse_upsert_row_request(
                     )
             request.fields_data.extend(struct_fields_data.values())
 
-        for _, field in enumerate(input_fields_info):
+        for field in input_fields_info:
             is_dynamic = False
             field_name = field["name"]
 
@@ -1551,20 +1559,10 @@ def search_requests_with_expr(
 
         json_type = kwargs.get(JSON_TYPE)
         if json_type is not None:
-            if json_type == DataType.INT8:
-                search_params[JSON_TYPE] = "Int8"
-            elif json_type == DataType.INT16:
-                search_params[JSON_TYPE] = "Int16"
-            elif json_type == DataType.INT32:
-                search_params[JSON_TYPE] = "Int32"
-            elif json_type == DataType.INT64:
-                search_params[JSON_TYPE] = "Int64"
-            elif json_type == DataType.BOOL:
-                search_params[JSON_TYPE] = "Bool"
-            elif json_type in (DataType.VARCHAR, DataType.STRING):
-                search_params[JSON_TYPE] = "VarChar"
-            else:
+            json_type_name = _JSON_TYPE_MAP.get(json_type)
+            if json_type_name is None:
                 raise ParamError(message=f"Unsupported json cast type: {json_type}")
+            search_params[JSON_TYPE] = json_type_name
 
         strict_cast = kwargs.get(STRICT_CAST)
         if strict_cast is not None:
@@ -1708,41 +1706,15 @@ def hybrid_search_request_with_ranker(
             ]
         )
 
-        if kwargs.get(RANK_GROUP_SCORER) is not None:
-            request.rank_params.extend(
-                [
-                    common_types.KeyValuePair(
-                        key=RANK_GROUP_SCORER, value=kwargs.get(RANK_GROUP_SCORER)
-                    )
-                ]
-            )
-
-        if kwargs.get(GROUP_BY_FIELD) is not None:
-            request.rank_params.extend(
-                [
-                    common_types.KeyValuePair(
-                        key=GROUP_BY_FIELD, value=utils.dumps(kwargs.get(GROUP_BY_FIELD))
-                    )
-                ]
-            )
-
-        if kwargs.get(GROUP_SIZE) is not None:
-            request.rank_params.extend(
-                [
-                    common_types.KeyValuePair(
-                        key=GROUP_SIZE, value=utils.dumps(kwargs.get(GROUP_SIZE))
-                    )
-                ]
-            )
-
-        if kwargs.get(STRICT_GROUP_SIZE) is not None:
-            request.rank_params.extend(
-                [
+        for param_key in (RANK_GROUP_SCORER, GROUP_BY_FIELD, GROUP_SIZE, STRICT_GROUP_SIZE):
+            val = kwargs.get(param_key)
+            if val is not None:
+                request.rank_params.append(
                     common_types.KeyValuePair(
-                        key=STRICT_GROUP_SIZE, value=utils.dumps(kwargs.get(STRICT_GROUP_SIZE))
+                        key=param_key,
+                        value=val if param_key == RANK_GROUP_SCORER else utils.dumps(val),
                     )
-                ]
-            )
+                )
 
         if isinstance(rerank, Function):
             request.function_score.CopyFrom(Prepare.ranker_to_function_score(rerank))