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

エンティティの Upsert

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

概要​

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

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

Q3LawAQIKht1FKbsM3EcoQAHnvc

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

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

マージモードでの Upsert​

マージモードを使用すると、他のフィールドを変更せずに、既存エンティティの特定のフィールドを更新できます。

NZNKwxm9ahmi87b487TcuCrNn4c

partial_update=True を設定し、主キーと更新するフィールドを指定します。

Zilliz Cloud は strong consistency のクエリで既存のエンティティを取得し、変更内容を保存済みのデータとマージして、マージ後のエンティティを挿入し、古いエンティティを削除します。

マージモードで既存のエンティティを更新する場合、autoID が有効であっても主キーは保持されます。主キーが存在しない場合、Zilliz Cloud は新しいエンティティの挿入を試みます。新しいエンティティを挿入するにはすべてのフィールドを指定する必要があり、そうでない場合、リクエストは missing-field エラーで失敗します。

部分更新が missing-field エラーで失敗した場合は、対象のエンティティが存在するかどうかを確認してください。既存のエンティティが存在しない場合、Zilliz Cloud は省略したフィールドの値を取得できません。

新しいエンティティには、insert または override モードでの upsert を使用します。個々のフィールドに対する後続の更新には、マージモードを使用します。

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

フィールド値の更新​

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

Upsert の動作：特記事項​

マージ機能を使用する前に検討しておくべき特記事項がいくつかあります。以下のケースでは、title と issue という 2 つのスカラーフィールド、主キー id、および vector というベクトルフィールドを持つコレクションを想定しています。

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

    issue フィールドが null になり得るとします。これらのフィールドを Upsert する際は、次の点に注意してください。

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

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

  • ダイナミックフィールドのキーの Upsert。

    例のコレクションでダイナミックキーを有効にしていて、あるエンティティのダイナミックフィールドのキーと値のペアが {"author": "John", "year": 2020, "tags": ["fiction"]} のようになっているとします。

    author、year、tags などのキーを含むエンティティを Upsert する場合、または他のキーを追加する場合は、次の点に注意してください。

    • partial_update を無効にして Upsert した場合のデフォルトの動作は override です。つまり、ダイナミックフィールドの値は、リクエストに含まれるスキーマで定義されていないすべてのフィールドとその値によって上書きされます。

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

    • partial_update を有効にして Upsert した場合のデフォルトの動作は merge です。つまり、ダイナミックフィールドの値は、リクエストに含まれるスキーマで定義されていないすべてのフィールドとその値とマージされます。

      たとえば、リクエストに含まれるデータが {"author": "John", "year": 2020, "tags": ["fiction"]} の場合、対象エンティティのダイナミックフィールドのキーと値のペアは、Upsert 後に {"author": "John", "year": 2020, "tags": ["fiction"], "genre": "fantasy"} になります。

  • JSON フィールドの Upsert。

    例のコレクションに、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 を実行する場合でも、構造体スキーマで定義されたすべてのサブフィールドを含む辞書のリストを指定する必要があります。

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

制限と制約​

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

  • upsert リクエストには、autoID が有効な場合でも常に対象エンティティの主キーを含める必要があります。autoID コレクションでは、主キーの扱いは Upsert モードによって異なります。

    • override モードでは、主キーは置き換える既存エンティティを特定するために使用され、Milvus は置き換え後のエンティティの新しい主キーを生成します。

    • マージモードでは、既存のエンティティを更新する場合も主キーは保持されます。主キーが存在しない場合、Zilliz Cloud は新しいエンティティの挿入を試みます。新しいエンティティを挿入するにはすべてのフィールドを指定する必要があり、そうでない場合、リクエストは missing-field エラーで失敗します。

  • 対象のコレクションはロード済みで、クエリに使用できる状態である必要があります。

  • リクエストで指定されたすべてのフィールドは、対象コレクションのスキーマに存在する必要があります。

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

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

コレクション内のエンティティの Upsert​

このセクションでは、my_collection という名前のコレクションにエンティティを Upsert します。このコレクションには、id、vector、title、issue という 2 つのフィールドしかありません。id フィールドは主フィールドで、title と issue フィールドはスカラーフィールドです。

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}

パーティション内のエンティティの Upsert​

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

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​

次の例では、my_collection 内の主キー 1 と 2 を持つエンティティの 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 フィールドを同時に更新した場合、後からの書き込みが先の書き込みを上書きする可能性があります。すべての同時変更を保持する必要がある場合は、アプリケーションレベルでの調整を行ってください。

例​

次の例では、主キー pk、ARRAY<VARCHAR> 型の tags フィールド、および embedding ベクトルフィールドを持つ小さな users コレクションを使用します。まず初期の 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 する際には、構造体スキーマで定義されたすべてのサブフィールドを含める必要があります。

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

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