Milvus Upsert not working as expected

[reenee]

Hi Team,

I’ve created a collection with the primary key feature enabled. However, when loading data from a CSV file using the upsert function, it doesn’t adhere to the primary key column. In cases where a record with the same ID already exists, instead of updating the existing entity, it’s creating a new entity with the same primary key.How can we resolve this issue?

Collection [id(primary_key_column), name, address]
collection.upsert(data_rows, primary_key_column=id)

[yhmo]

An example to show upsert(). It inserts duplicate IDs [0, 1, 2, 3, 4] by upsert().
After 1000 entities are upserted, the collection.num_entiites returns 1000 because this interface only calculates inserted rows. The "count(*)" returns 5 which is correct. Finally there are only 5 entities in the collection, the others were overwritten by the same primary key.

import random

from pymilvus import (
    connections,
    FieldSchema, CollectionSchema, DataType,
    Collection,
    utility,
)


connections.connect(host='localhost', port='19530')
print(utility.get_server_version())

collection_name = "test"
dim = 128
metric_type = "L2"

# create collection
id_field = FieldSchema(
    name="id",
    dtype=DataType.INT64,
    is_primary=True,
    auto_id=False,
)
vector_field = FieldSchema(
    name="vector",
    dtype = DataType.FLOAT_VECTOR,
    dim=dim
)

name_field = FieldSchema(
    name="name",
    dtype = DataType.VARCHAR,
    max_length=1000,
)

address_field = FieldSchema(
    name="address",
    dtype = DataType.VARCHAR,
    max_length=1000,
)

schema = CollectionSchema(fields=[id_field, vector_field, name_field, address_field])

if utility.has_collection(collection_name):
    utility.drop_collection(collection_name)

collection = Collection(name=collection_name, schema=schema)
print(f"Collection '{collection_name}' created")
print(collection.describe())

for i in range(1000):
    collection.upsert({"id": i%5, "vector": [random.random() for _ in range(dim)], "name": f"name_{i}", "address": f"address_{i}"})

collection.flush()
print("collection.num_entities returns row count", collection.num_entities)

index_params = {
    'metric_type': metric_type,
    'index_type': "FLAT",
    'params': {},
}
collection.create_index(field_name="vector", index_params=index_params)
print("index created")

collection.load()
print("collection loaded")

res = collection.query(expr="", consistency_level="Strong", output_fields=["count(*)"])
print("count(*) returns row count", res)

res = collection.query(expr="id == 3", output_fields=["name", "address"])
for r in res:
    print(r)

This script outputs these lines:

v2.3.2-dev
Collection 'test' created
{'collection_name': 'test', 'auto_id': False, 'num_shards': 1, 'description': '', 'fields': [{'field_id': 100, 'name': 'id', 'description': '', 'type': 5, 'params': {}, 'element_type': 0, 'is_primary': True}, {'field_id': 101, 'name': 'vector', 'description': '', 'type': 101, 'params': {'dim': 128}, 'element_type': 0}, {'field_id': 102, 'name': 'name', 'description': '', 'type': 21, 'params': {'max_length': 1000}, 'element_type': 0}, {'field_id': 103, 'name': 'address', 'description': '', 'type': 21, 'params': {'max_length': 1000}, 'element_type': 0}], 'aliases': [], 'collection_id': 445921815684858672, 'consistency_level': 2, 'properties': {}, 'num_partitions': 1}
collection.num_entities returns row count 1000
index created
collection loaded
count(*) returns row count [{'count(*)': 5}]
{'id': 3, 'name': 'name_998', 'address': 'address_998'}