跳到主要内容
版本：v2.6.x

Upsert Entity

upsert 操作为在 Collection 中插入或更新实体提供了一种便捷的方法。

操作概述​

根据 upsert 请求中提供的主键是否存在于 Collection 中，您可以使用 upsert 插入新实体或更新现有实体。如果找不到主键，则进行插入操作。否则，将执行更新操作。

在 Milvus 中，upsert 可在 覆盖 或 合并 模式下工作。

覆盖模式下的 upsert​

覆盖模式下的 upsert 请求结合了插入和删除操作。当收到针对现有实体的 upsert 请求时，Milvus 会插入请求有效载荷中携带的数据，并同时删除数据中指定原始主键的现有实体。

Upsert In Override Mode

如果目标 Collection 的主键字段已启用 autoID，则 upsert 请求仍必须包括目标实体的主键。Milvus 使用所提供的主键来定位要替换的实体，并在插入请求有效载荷中携带的数据前生成一个新的主键。

对于已启用 nullable 的字段，如果不需要更新，可以在 upsert 请求中省略。

在合并模式下 upsert​

您还可以使用 partial_update 标志，使 upsert 请求以合并模式运行。这样就可以在请求有效载荷中只包含需要更新的字段。

Upsert In Merge Mode

要执行合并，请在 upsert 请求中将 partial_update 设置为 True，并将主键和需要更新的字段设置为新值。

收到这样的请求后，Milvus 会执行 Strong Consistency 查询以检索实体，根据请求中的数据更新字段值，插入修改后的数据，然后用请求中携带的原始主键删除现有实体。

对于 ARRAY 字段，合并模式支持两种运算符：ARRAY_APPEND 和 ARRAY_REMOVE。这些运算符可让您在现有 ARRAY 字段中追加元素或删除匹配元素，而无需先查询实体以获取其当前值。有关详细信息，请参阅 使用部分更新运算符的 Upsert ARRAY 字段。

Upsert 行为：特别注意事项​

在使用合并功能之前，有几个特别注意事项需要考虑。以下情况假定您有一个 Collection，其中有两个标量字段，分别名为 title 和 issue，以及一个主键 id 和一个向量字段 vector。

  • 启用 nullable 的 upsert 字段 。

假设 issue 字段可以为空。在 upsert 这些字段时，请注意以下几点：

  • 如果在 upsert 请求中省略 issue 字段并禁用 partial_update，issue 字段将更新为 null，而不是保留其原始值。

  • 要保留 issue 字段的原始值，要么启用 partial_update 并省略 issue 字段，要么在 upsert 请求中包含 issue 字段及其原始值。

  • 在动态字段中 upsert 键。

假设在示例 Collection 中启用了动态键，实体动态字段中的键值对与 {"author": "John", "year": 2020, "tags": ["fiction"]} 类似。

当你 upsert 实体的键，如 author,year, 或 tags，或添加其他键时，请注意：

  • 如果 upsert partial_update 时禁用，默认行为是 覆盖。这意味着动态字段的值将被请求中包含的所有非 Schema 定义的字段及其值覆盖。

例如，如果请求中包含的数据是 {"author": "Jane", "genre": "fantasy"}，目标实体动态字段中的键值对将更新为该值。

  • 如果在启用 partial_update 的情况下进行 upsert，默认行为是 合并。这意味着动态字段的值将与请求中包含的所有非 Schema 定义的字段及其值合并。

例如，如果请求中包含的数据是 {"author": "John", "year": 2020, "tags": ["fiction"]}，则目标实体动态字段中的键值对在 upsert 后将变成 {"author": "John", "year": 2020, "tags": ["fiction"], "genre": "fantasy"}。

  • upsert 一个 JSON 字段。

假设示例 Collection 有一个名为 extras 的 Schema 定义 JSON 字段，实体的此 JSON 字段中的键值对类似于 {"author": "John", "year": 2020, "tags": ["fiction"]}。

当您使用修改后的 JSON 数据 upsert 实体的 extras 字段时，请注意该 JSON 字段被视为一个整体，您不能有选择地更新单个键。换句话说，JSON 字段 不 支持 合并 模式下的 upsert。

  • upsert ARRAY 字段。

在合并模式下，ARRAY 字段支持 ARRAY_APPEND 和 ARRAY_REMOVE 部分更新运算符。当您想在现有 ARRAY 字段中添加元素或移除匹配元素而不替换整个数组值时，请使用这些运算符。

限制和约束​

根据上述内容，有几个限制和约束需要遵循：

  • upsert 请求必须始终包含目标实体的主键，即使启用了 autoID。对于 autoID Collection，请求中的主键标识了要替换的现有实体。Milvus 会为插入的替换实体生成新的主键。

  • 目标 Collection 必须已加载并可供查询。

  • 请求中指定的所有字段必须存在于目标 Collection 的 Schema 中。

  • 请求中指定的所有字段的值必须与 Schema 中定义的数据类型相匹配。

  • 对于使用函数从另一个字段派生出来的任何字段，Milvus 将在 upsert 过程中删除派生字段，以便重新计算。

upsert Collection 中的 Entity​

在本节中，我们将把实体 upsert 到名为 my_collection 的 Collection 中。该 Collection 只有两个字段，分别名为 id,vector,title 和 issue。id 字段是主键字段，而 title 和 issue 字段是标量字段。

这三个实体如果存在于 Collection 中，将被包含 upsert 请求的实体覆盖。

Python
from pymilvus import MilvusClient

client = MilvusClient(
    uri="http://localhost:19530",
    token="root:Milvus"
)

data=[
    {
        "id": 0,
        "vector": [-0.619954382375778, 0.4479436794798608, -0.17493894838751745, -0.4248030059917294, -0.8648452746018911],
        "title": "Artificial Intelligence in Real Life",
        "issue": "vol.12"
    }, {
        "id": 1,
        "vector": [0.4762662251462588, -0.6942502138717026, -0.4490002642657902, -0.628696575798281, 0.9660395877041965],
        "title": "Hollow Man",
        "issue": "vol.19"
    }, {
        "id": 2,
        "vector": [-0.8864122635045097, 0.9260170474445351, 0.801326976181461, 0.6383943392381306, 0.7563037341572827],
        "title": "Treasure Hunt in Missouri",
        "issue": "vol.12"
    }
]

res = client.upsert(
    collection_name='my_collection',
    data=data
)

print(res)

# Output
# {'upsert_count': 3}

upsert Partition 中的 Entity​

您还可以将实体 upsert 到指定的 Partition 中。以下代码片段假定你的 Collection 中有一个名为 PartitionA 的 Partition。

如果 Partition 中存在三个实体，它们将被请求中包含的实体覆盖。

Python
data=[
    {
        "id": 10,
        "vector": [0.06998888224297328, 0.8582816610326578, -0.9657938677934292, 0.6527905683627726, -0.8668460657158576],
        "title": "Layour Design Reference",
        "issue": "vol.34"
    },
    {
        "id": 11,
        "vector": [0.6060703043917468, -0.3765080534566074, -0.7710758854987239, 0.36993888322346136, 0.5507513364206531],
        "title": "Doraemon and His Friends",
        "issue": "vol.2"
    },
    {
        "id": 12,
        "vector": [-0.9041813104515337, -0.9610546012461163, 0.20033003106083358, 0.11842506351635174, 0.8327356724591011],
        "title": "Pikkachu and Pokemon",
        "issue": "vol.12"
    },
]

res = client.upsert(
    collection_name="my_collection",
    data=data,
    partition_name="partitionA"
)

print(res)

# Output
# {'upsert_count': 3}

在合并模式下 Upsert Entity​

下面的代码示例演示了如何通过部分更新来 upsert 实体。只提供需要更新的字段及其新值，以及显式部分更新标记。

在下面的示例中，upsert 请求中指定的实体的 issue 字段将更新为请求中包含的值。

说明

在合并模式下执行 upsert 时，请确保请求中涉及的实体具有相同的字段集。假设有两个或更多实体要进行 upsert，如以下代码片段所示，它们必须包含相同的字段，以防止出现错误并保持数据完整性。

Python
data=[
    {
        "id": 1,
        "issue": "vol.14"
    },
    {
        "id": 2,
        "issue": "vol.7"
    }
]

res = client.upsert(
    collection_name="my_collection",
    data=data,
    partial_update=True
)

print(res)

# Output
# {'upsert_count': 2}

使用部分更新运算符 upsert ARRAY 字段​

在引入部分更新运算符之前，更新 ARRAY 字段的部分内容需要客户端读取-修改-写入流程：查询现有数组，在应用代码中进行修改，然后 upsert 完整的替换值。部分更新运算符可以让你只发送要追加或移除的元素，从而减少了客户端逻辑，避免了 upsert 前的额外读取。

假设主键为 1 的实体已经有 tags = ["new", "trial"]。如果没有部分更新运算符，向数组中添加 "premium" 就需要 upsert 完整的替换数组：

Python
client.upsert(
    collection_name="users",
    data=[{"pk": 1, "tags": ["new", "trial", "premium"]}],
    partial_update=True,
)

使用 ARRAY_APPEND 时，只需发送要添加的元素：

Python
from pymilvus import FieldOp

client.upsert(
    collection_name="users",
    data=[{"pk": 1, "tags": ["premium"]}],
    field_ops={"tags": FieldOp.array_append()},
)
说明

通过 field_ops 将任一运算符附加到字段，都会隐式启用部分更新语义。因此，您 无需 在传递 partial_update=True 的同时传递 field_ops。

限制​

  • 有效负载值必须与 ARRAY 目标字段的 element_type 匹配。例如，如果目标字段是 ARRAY<VARCHAR>，则有效载荷必须包含字符串值。
  • ARRAY_APPEND 和 ARRAY_REMOVE 支持 ARRAY 字段，其 element_type 为 BOOL,INT8,INT16,INT32,INT64,FLOAT,DOUBLE, 或 VARCHAR。
  • 进行 ARRAY_APPEND 操作后，生成的数组长度不得超过字段的 max_capacity。
  • 对同一实体的并发 upsert 在不同请求之间不是原子的。如果两个请求同时更新同一个 ARRAY 字段，后一个写入会覆盖前一个写入。如果需要保留所有并发更改，请使用应用程序级协调。

示例​

下面的示例使用了一个小型 users Collection，其中有一个主键 pk、一个 tags 类型的字段 ARRAY<VARCHAR> 和一个 embedding 向量字段。它首先插入两个具有初始 tags 值的实体，然后使用 ARRAY_APPEND 和 ARRAY_REMOVE 来展示每个运算符如何更改存储的数组。

Python
from pymilvus import DataType, FieldOp, MilvusClient

client = MilvusClient(
    uri="http://localhost:19530",
    token="root:Milvus"
)

# 1. Create a collection with an ARRAY<VARCHAR> field
schema = client.create_schema(enable_dynamic_field=False)
schema.add_field("pk", DataType.INT64, is_primary=True)
schema.add_field("embedding", DataType.FLOAT_VECTOR, dim=5)
schema.add_field(
    "tags",
    DataType.ARRAY,
    element_type=DataType.VARCHAR,
    max_capacity=8,
    max_length=32,
)

index_params = client.prepare_index_params()
index_params.add_index(
    field_name="embedding",
    index_type="AUTOINDEX",
    metric_type="L2",
)

client.create_collection(
    collection_name="users",
    schema=schema,
    index_params=index_params
)

# 2. Seed two entities
client.insert(
    collection_name="users",
    data=[
        {"pk": 1, "embedding": [0.1, 0.2, 0.3, 0.4, 0.5], "tags": ["new"]},
        {"pk": 2, "embedding": [0.6, 0.7, 0.8, 0.9, 1.0], "tags": ["new", "trial"]},
    ],
)

# 3. Append tags without reading the existing ARRAY values
client.upsert(
    collection_name="users",
    data=[
        {"pk": 1, "tags": ["premium", "vip"]},
        {"pk": 2, "tags": ["premium"]},
    ],
    field_ops={"tags": FieldOp.array_append()},
)

res = client.query(
    collection_name="users",
    filter="pk in [1, 2]",
    output_fields=["pk", "tags"],
)
print(res)

# Example output:
# data: [
#   "{'pk': 1, 'tags': ['new', 'premium', 'vip']}",
#   "{'pk': 2, 'tags': ['new', 'trial', 'premium']}"
# ]

# 4. Remove matching tags without replacing the full ARRAY field
client.upsert(
    collection_name="users",
    data=[
        {"pk": 1, "tags": ["new"]},
        {"pk": 2, "tags": ["trial"]},
    ],
    field_ops={"tags": FieldOp.array_remove()},
)

res = client.query(
    collection_name="users",
    filter="pk in [1, 2]",
    output_fields=["pk", "tags"],
)
print(res)

# Example output:
# data: [
#   "{'pk': 1, 'tags': ['premium', 'vip']}",
#   "{'pk': 2, 'tags': ['new', 'premium']}"
# ]