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

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

マージモードで既存のエンティティを更新すると、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 フィールドを省略するか、元の値を指定した issue フィールドを upsert リクエストに含める必要があります。

  • 動的フィールド内のキーの 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 を実行する場合でも、struct スキーマで定義されたすべてのサブフィールドをそれぞれに含む辞書のリストを指定する必要があります。

    詳細については、マージモードでの 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 フィールドの一部を更新するにはクライアント側での read-modify-write フローが必要でした。つまり、既存の配列をクエリし、アプリケーションコードで変更し、完全な置き換え値を 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 する際には、struct スキーマで定義されたすべてのサブフィールドを含める必要があります。

以下の例では、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