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

エンティティの Upsert

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

概要​

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

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

Q3LawAQIKht1FKbsM3EcoQAHnvc

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

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

マージモードでの Upsert​

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

NZNKwxm9ahmi87b487TcuCrNn4c

partial_update=True を設定し、プライマリキーと更新するフィールドを指定します。

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

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

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

新しいエンティティには、insert または上書きモードの 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 フィールドを省略するか、元の値を指定した issue フィールドを upsert リクエストに含める必要があります。

  • 動的フィールドのキーを Upsert する。

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

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

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

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

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

      たとえば、リクエストに含まれるデータが {"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 フィールドは マージ モードでの 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 を使用するコレクションでは、プライマリキーの扱いは Upsert モードによって異なります。

    • 上書きモードでは、プライマリキーは置き換える既存のエンティティを識別し、Milvus は置き換え後のエンティティ用に新しいプライマリキーを生成します。

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

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

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

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

  • 関数を使用して他のフィールドから派生したフィールドについては、再計算できるように、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
from pymilvus import FieldOp

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 する際は、struct スキーマで定義されたすべてのサブフィールドを含める必要があります。

以下の例では、6 つのサブフィールドを持つ StructArray フィールドである chunks フィールドをマージモードで Upsert する方法を示します。操作が完了すると、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