fix:use list join instead of += to operate bytes to prevent oom (#3181)

use list join instead of += to operate bytes to prevent oom
see #3019

---------

Signed-off-by: silas.jiang <silas.jiang@zilliz.com>
Co-authored-by: silas.jiang <silas.jiang@zilliz.com>

Author: jac0626

pymilvus/client/entity_helper.py

@@ -356,8 +356,37 @@ def entity_to_array_arr(entity_values: List[Any], field_info: Any):
     return convert_to_array_arr(entity_values, field_info)
 
 
+def flush_vector_bytes(
+    field_data: schema_types.FieldData, vector_bytes_cache: Dict[int, List[bytes]]
+):
+    """Flush the temporary byte list for bytes vector fields, merging all collected bytes.
+
+    This function is used to optimize performance by avoiding O(n²) memory operations
+    caused by using += operations in pack_field_value_to_field_data.
+    Supports: INT8_VECTOR, BINARY_VECTOR, FLOAT16_VECTOR, BFLOAT16_VECTOR
+    """
+    field_id = id(field_data)
+    bytes_list = vector_bytes_cache.pop(field_id, None)
+    if not bytes_list:
+        return
+
+    vector_attr_map = {
+        DataType.INT8_VECTOR: "int8_vector",
+        DataType.BINARY_VECTOR: "binary_vector",
+        DataType.FLOAT16_VECTOR: "float16_vector",
+        DataType.BFLOAT16_VECTOR: "bfloat16_vector",
+    }
+
+    attr_name = vector_attr_map.get(field_data.type)
+    if attr_name:
+        setattr(field_data.vectors, attr_name, b"".join(bytes_list))
+
+
 def pack_field_value_to_field_data(
-    field_value: Any, field_data: schema_types.FieldData, field_info: Any
+    field_value: Any,
+    field_data: schema_types.FieldData,
+    field_info: Any,
+    vector_bytes_cache: Dict[int, List[bytes]],
 ):
     field_type = field_data.type
     field_name = field_info["name"]
@@ -462,7 +491,12 @@ def pack_field_value_to_field_data(
                     field_data.vectors.dim = field_info.get("params", {}).get("dim", 0)
             else:
                 field_data.vectors.dim = len(field_value) * 8
-                field_data.vectors.binary_vector += bytes(field_value)
+                b_bytes = bytes(field_value)
+
+                field_id = id(field_data)
+                if field_id not in vector_bytes_cache:
+                    vector_bytes_cache[field_id] = []
+                vector_bytes_cache[field_id].append(b_bytes)
         except (TypeError, ValueError) as e:
             raise DataNotMatchException(
                 message=ExceptionsMessage.FieldDataInconsistent
@@ -489,7 +523,11 @@ def pack_field_value_to_field_data(
                     )
 
                 field_data.vectors.dim = len(v_bytes) // 2
-                field_data.vectors.float16_vector += v_bytes
+
+                field_id = id(field_data)
+                if field_id not in vector_bytes_cache:
+                    vector_bytes_cache[field_id] = []
+                vector_bytes_cache[field_id].append(v_bytes)
         except (TypeError, ValueError) as e:
             raise DataNotMatchException(
                 message=ExceptionsMessage.FieldDataInconsistent
@@ -516,7 +554,11 @@ def pack_field_value_to_field_data(
                     )
 
                 field_data.vectors.dim = len(v_bytes) // 2
-                field_data.vectors.bfloat16_vector += v_bytes
+
+                field_id = id(field_data)
+                if field_id not in vector_bytes_cache:
+                    vector_bytes_cache[field_id] = []
+                vector_bytes_cache[field_id].append(v_bytes)
         except (TypeError, ValueError) as e:
             raise DataNotMatchException(
                 message=ExceptionsMessage.FieldDataInconsistent
@@ -562,7 +604,11 @@ def pack_field_value_to_field_data(
                     )
 
                 field_data.vectors.dim = len(i_bytes)
-                field_data.vectors.int8_vector += i_bytes
+
+                field_id = id(field_data)
+                if field_id not in vector_bytes_cache:
+                    vector_bytes_cache[field_id] = []
+                vector_bytes_cache[field_id].append(i_bytes)
         except (TypeError, ValueError) as e:
             raise DataNotMatchException(
                 message=ExceptionsMessage.FieldDataInconsistent

pymilvus/client/prepare.py

@@ -761,6 +761,10 @@ def _parse_row_request(
         }
         field_info_map = {field["name"]: field for field in input_fields_info}
 
+        # Local cache for temporary byte lists for bytes vector fields
+        # key: field_data object id, value: list of bytes
+        vector_bytes_cache: Dict[int, List[bytes]] = {}
+
         (
             struct_fields_data,
             struct_info_map,
@@ -799,7 +803,9 @@ def _parse_row_request(
                             "default_value", None
                         ):
                             field_data.valid_data.append(v is not None)
-                        entity_helper.pack_field_value_to_field_data(v, field_data, field_info)
+                        entity_helper.pack_field_value_to_field_data(
+                            v, field_data, field_info, vector_bytes_cache
+                        )
                     elif k in struct_fields_data:
                         # Array of structs format
                         try:
@@ -823,7 +829,9 @@ def _parse_row_request(
                     field_info, field_data = field_info_map[key], fields_data[key]
                     if field_info.get("nullable", False) or field_info.get("default_value", None):
                         field_data.valid_data.append(False)
-                        entity_helper.pack_field_value_to_field_data(None, field_data, field_info)
+                        entity_helper.pack_field_value_to_field_data(
+                            None, field_data, field_info, vector_bytes_cache
+                        )
                     else:
                         raise DataNotMatchException(
                             message=ExceptionsMessage.InsertMissedField % key
@@ -852,6 +860,16 @@ def _parse_row_request(
                     struct_sub_fields_data[struct_name][field_name]
                 )
 
+        # Flush all bytes vector field temporary byte lists to optimize memory usage
+        for field_data in fields_data.values():
+            if field_data.type in (
+                DataType.INT8_VECTOR,
+                DataType.BINARY_VECTOR,
+                DataType.FLOAT16_VECTOR,
+                DataType.BFLOAT16_VECTOR,
+            ):
+                entity_helper.flush_vector_bytes(field_data, vector_bytes_cache)
+
         request.fields_data.extend(fields_data.values())
         request.fields_data.extend(struct_fields_data.values())
 
@@ -889,6 +907,10 @@ def _parse_upsert_row_request(
         field_info_map = {field["name"]: field for field in input_fields_info}
         field_len = {field["name"]: 0 for field in input_fields_info}
 
+        # Local cache for temporary byte lists for bytes vector fields
+        # key: field_data object id, value: list of bytes
+        vector_bytes_cache: Dict[int, List[bytes]] = {}
+
         # Use common struct data setup (only if not partial update)
         if partial_update:
             struct_fields_data = {}
@@ -936,7 +958,9 @@ def _parse_upsert_row_request(
                             "default_value", None
                         ):
                             field_data.valid_data.append(v is not None)
-                        entity_helper.pack_field_value_to_field_data(v, field_data, field_info)
+                        entity_helper.pack_field_value_to_field_data(
+                            v, field_data, field_info, vector_bytes_cache
+                        )
                         field_len[k] += 1
                     elif k in struct_fields_data:
                         # Handle struct field (array of structs)
@@ -966,7 +990,9 @@ def _parse_upsert_row_request(
                     if field_info.get("nullable", False) or field_info.get("default_value", None):
                         field_data.valid_data.append(False)
                         field_len[key] += 1
-                        entity_helper.pack_field_value_to_field_data(None, field_data, field_info)
+                        entity_helper.pack_field_value_to_field_data(
+                            None, field_data, field_info, vector_bytes_cache
+                        )
                     else:
                         raise DataNotMatchException(
                             message=ExceptionsMessage.InsertMissedField % key
@@ -998,6 +1024,16 @@ def _parse_upsert_row_request(
         fields_data = {k: v for k, v in fields_data.items() if field_len[k] > 0}
         request.fields_data.extend(fields_data.values())
 
+        # Flush all bytes vector field temporary byte lists to optimize memory usage
+        for field_data in fields_data.values():
+            if field_data.type in (
+                DataType.INT8_VECTOR,
+                DataType.BINARY_VECTOR,
+                DataType.FLOAT16_VECTOR,
+                DataType.BFLOAT16_VECTOR,
+            ):
+                entity_helper.flush_vector_bytes(field_data, vector_bytes_cache)
+
         if struct_fields_data:
             # reconstruct the struct array fields data (same as in insert)
             for struct in input_struct_field_info: