メインコンテンツまでスキップ

エンティティの Upsert

upsert 操作は、collection 内のエンティティを挿入または更新するための便利な方法を提供します。

Overview​

upsert は、upsert リクエストで指定された主キーが collection 内に存在するかどうかに応じて、新しいエンティティを挿入することも、既存のエンティティを更新することもできます。主キーが見つからない場合は、insert 操作が実行されます。そうでない場合は、update 操作が実行されます。

upsert リクエストは、insert と delete を組み合わせたものです。既存のエンティティに対する upsert リクエストを受信すると、Zilliz Cloud はリクエストペイロードに含まれるデータを挿入し、同時にデータ内で指定された元の主キーを持つ既存エンティティを削除します。

Q3LawAQIKht1FKbsM3EcoQAHnvc

対象の collection で主フィールドに autoID が有効になっている場合でも、upsert リクエストには対象エンティティの主キーを含める必要があります。Zilliz Cloud は提供された主キーを使用して置き換えるエンティティを特定し、挿入前にリクエストペイロード内のデータに対して新しい主キーを生成します。

nullable が有効になっているフィールドについては、更新が不要であれば upsert リクエスト内で省略できます。

マージモードでの Upsert​

partial_update フラグを使用して、upsert リクエストをマージモードで動作させることもできます。これにより、リクエストペイロードには更新が必要なフィールドのみを含めることができます。

NZNKwxm9ahmi87b487TcuCrNn4c

マージを実行するには、upsert リクエストで partial_update を True に設定し、主キーと更新するフィールドおよびその新しい値を一緒に指定します。

このようなリクエストを受信すると、Zilliz Cloud は strong consistency で query を実行してエンティティを取得し、リクエスト内のデータに基づいてフィールド値を更新し、変更されたデータを挿入した後、リクエストに含まれる元の主キーを持つ既存エンティティを削除します。

ARRAY フィールドについては、マージモードで ARRAY_APPEND と ARRAY_REMOVE の 2 つの演算子をサポートしています。これらの演算子を使うと、エンティティを最初に query して現在の値を取得しなくても、既存の ARRAY フィールドに要素を追加したり、一致する要素を削除したりできます。詳細については、部分更新演算子を使った ARRAY フィールドの Upsert を参照してください。

フィールド値の更新​

既存エンティティのフィールド値を更新するには、マージモードでの upsert を使用します。このモードでは、リクエストに含まれるフィールドのみが更新され、その他のすべてのフィールドは既存の値を保持します。

Upsert の動作: 特記事項​

マージ機能を使用する前に考慮すべき特記事項がいくつかあります。以下のケースでは、title と issue という名前の 2 つの scalar フィールド、id という主キー、そして vector という vector フィールドを持つ collection があると仮定しています。

  • nullable が有効なフィールドの Upsert。

    issue フィールドが null を許容するとします。これらのフィールドを upsert する場合、次の点に注意してください。

    • upsert リクエストで issue フィールドを省略し、partial_update を無効にすると、issue フィールドは元の値を保持するのではなく null に更新されます。

    • issue フィールドの元の値を保持するには、partial_update を有効にして issue フィールドを省略するか、upsert リクエストに issue フィールドを元の値付きで含める必要があります。

  • dynamic field 内のキーの Upsert。

    サンプル collection で dynamic key を有効にしており、あるエンティティの dynamic field のキーと値のペアが {"author": "John", "year": 2020, "tags": ["fiction"]} のようなものだとします。

    author、year、tags のようなキーでエンティティを upsert したり、他のキーを追加したりする場合は、次の点に注意してください。

    • partial_update を無効にして upsert する場合、デフォルトの動作は override です。これは、dynamic field の値が、リクエストに含まれるスキーマ未定義のすべてのフィールドとその値で上書きされることを意味します。

      たとえば、リクエストに含まれるデータが {"author": "Jane", "genre": "fantasy"} の場合、対象エンティティの dynamic field 内のキーと値のペアはそれに更新されます。

    • partial_update を有効にして upsert する場合、デフォルトの動作は merge です。これは、dynamic field の値が、リクエストに含まれるスキーマ未定義のすべてのフィールドとその値とマージされることを意味します。

      たとえば、リクエストに含まれるデータが {"author": "John", "year": 2020, "tags": ["fiction"]} の場合、upsert 後に dynamic field 内のキーと値のペアは {"author": "John", "year": 2020, "tags": ["fiction"], "genre": "fantasy"} になります。

  • JSON フィールドの Upsert。

    サンプル collection に extras というスキーマ定義済みの JSON フィールドがあり、あるエンティティのこの JSON フィールド内のキーと値のペアが {"author": "John", "year": 2020, "tags": ["fiction"]} のようなものだとします。

    修正された JSON データでエンティティの extras フィールドを upsert する場合、JSON フィールドは全体として扱われ、個々のキーを選択的に更新することはできません。言い換えると、JSON フィールドは merge モードでの upsert をサポートしません。

  • ARRAY フィールドの Upsert。

    デフォルトでは、マージモードの ARRAY フィールドは REPLACE セマンティクスに従います。つまり、リクエストに含まれる値が既存の配列を上書きします。より細かい更新のために、Zilliz Cloud は次の 2 つの演算子もサポートしています。

    • ARRAY_APPEND は、リクエストペイロード内の要素を既存の配列に追加します。

    • ARRAY_REMOVE は、既存の配列から、リクエストペイロード内の値と一致するすべての要素を削除します。

    演算子の構文、サポートされる要素型、その他の制約については、部分更新演算子を使った array フィールドの Upsert を参照してください。

  • StructArray フィールドの Upsert。

    エンティティ内の StructArray フィールドを upsert すると、そのフィールド値は上書きされます。これを行うには、マージモードで upsert を実行する場合でも、struct スキーマで定義されたすべてのサブフィールドを含む辞書のリストを指定する必要があります。

    詳細については、マージモードでの StructArray フィールドの Upsert を参照してください。

制限事項​

上記の内容に基づき、従うべき制限事項がいくつかあります。

  • upsert リクエストには、autoID が有効な場合でも、常に対象エンティティの主キーを含める必要があります。autoID collection では、リクエスト内の主キーは置き換える既存エンティティを識別します。Milvus は挿入される置換エンティティに対して新しい主キーを生成します。

  • 対象の collection は load 済みで、query に利用可能である必要があります。

  • リクエストで指定されるすべてのフィールドは、対象 collection のスキーマ内に存在している必要があります。

  • リクエストで指定されるすべてのフィールドの値は、スキーマで定義されたデータ型と一致している必要があります。

  • 関数を使って他のフィールドから派生したフィールドについては、再計算を可能にするため、Zilliz Cloud は upsert 中にその派生フィールドを削除します。

collection 内のエンティティの Upsert​

このセクションでは、my_collection という名前の collection にエンティティを upsert します。この collection には、id、vector、title、issue という 2 つのフィールドのみがあります。id フィールドは主フィールドであり、title と issue フィールドは scalar フィールドです。

collection 内にこれら 3 つのエンティティが存在する場合、upsert リクエストに含まれるものによって上書きされます。

python
from pymilvus import MilvusClient

client = MilvusClient(
    uri="YOUR_CLUSTER_ENDPOINT",
    token="YOUR_CLUSTER_TOKEN"
)

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}

partition 内のエンティティの Upsert​

指定した partition にエンティティを upsert することもできます。以下のコードスニペットでは、collection 内に PartitionA という名前の partition があることを前提としています。

partition 内にこれら 3 つのエンティティが存在する場合、リクエストに含まれるものによって上書きされます。

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 する​

以下のコード例は、部分更新を使ってエンティティを upsert する方法を示しています。更新が必要なフィールドとその新しい値だけを、明示的な部分更新フラグとともに指定します。

次の例では、upsert リクエストで指定されたエンティティの issue フィールドが、リクエストに含まれる値に更新されます。

📘Notes

マージモードで upsert を実行する際は、リクエストに含まれるエンティティが同じフィールドセットを持っていることを確認してください。以下のコードスニペットのように upsert されるエンティティが 2 つ以上ある場合、エラーを防ぎ、データ整合性を維持するために、それらが同一のフィールドを含んでいることが重要です。

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}

マージモードで ARRAY フィールドを Upsert する​

部分更新演算子（ARRAY_APPEND と ARRAY_REMOVE）が導入される前は、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
client.upsert(
    collection_name="users",
    data=[{"pk": 1, "tags": ["premium"]}],
    field_ops={"tags": FieldOp.array_append()},
)
📘Notes

field_ops を介していずれかの演算子をフィールドに適用すると、部分更新のセマンティクスが暗黙的に有効になります。したがって、field_ops と一緒に partial_update=True を渡す必要はありません。

制限事項​

  • ペイロード値は、対象の ARRAY フィールドの element_type と一致している必要があります。たとえば、対象フィールドが ARRAY<VARCHAR> の場合、ペイロードには文字列値を含める必要があります。

  • このリリースでは、ARRAY_APPEND と ARRAY_REMOVE は、element_type が BOOL、INT8、INT16、INT32、INT64、FLOAT、DOUBLE、または VARCHAR の ARRAY フィールドをサポートします。

  • ARRAY_APPEND 操作後、結果の配列長はそのフィールドの max_capacity を超えてはなりません。

  • 同じエンティティに対する同時 upsert は、リクエスト間でアトミックではありません。2 つのリクエストが同じ ARRAY フィールドを同時に更新した場合、後からの書き込みが先の書き込みを上書きすることがあります。すべての同時変更を保持する必要がある場合は、アプリケーションレベルで調整を行ってください。

例​

以下の例では、小さな users collection を使用します。この collection には、主キー pk、ARRAY<VARCHAR> 型の tags フィールド、および embedding vector フィールドがあります。まず初期 tags 値を持つ 2 つのエンティティを挿入し、その後 ARRAY_APPEND と ARRAY_REMOVE を使って、各演算子が保存済み配列をどのように変更するかを示します。

python
from pymilvus import DataType, FieldOp, MilvusClient

client = MilvusClient(
    uri="YOUR_CLUSTER_ENDPOINT",
    token="YOUR_CLUSTER_TOKEN"
)

# 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']}"
# ]

マージモードで StructArray フィールドを Upsert する​

エンティティ内の StructArray フィールドを upsert すると、そのフィールド値は上書きされます。つまり、StructArray フィールドを upsert する際には、struct スキーマで定義されているすべてのサブフィールドを含める必要があります。

以下の例は、マージモードで chunks フィールドを upsert する方法を示しています。これは 6 つのサブフィールドを持つ StructArray フィールドです。操作が完了すると、id 1 のエンティティの chunks フィールドは、リクエストで指定された 2 要素の struct 配列に設定されます。

python
client.upsert(
    collection_name="books",
    data=[{
        "id": 1,
        "chunks": [
            {
              "text": "Use HNSW efSearch to trade recall for latency.",
              "section": "index",
              "page": 1,
              "quality_score": 0.92,
              "has_code": True,
              "emb_list_vector": [0.11, 0.21, 0.31, 0.41]
            },
            {
              "text": "Range search returns vectors within a distance boundary.",
              "section": "search",
              "page": 2,
              "quality_score": 0.86,
              "has_code": False,
              "emb_list_vector": [0.18, 0.23, 0.29, 0.36]
            }
        ]
    }],
    partial_update=True
)
Ctrl I