Azure Cosmos DB - UPSERT

前提

Partition Key は/pk、Unique keys は/col1, /col2とする。

UPSERT

使用コード

UPSERT には SDK の upsert を使用する。

const { CosmosClient } = require('@azure/cosmos');
const config = require('./config');

async function upsert() {
  const { connectionString, databaseId, containerId } = config;

  const client = new CosmosClient(connectionString);
  const database = client.database(databaseId);
  const container = database.container(containerId);

  try {
    const item = {
      // 内容は適時変更
      pk: 'A',
      id: '0001',
      col1: 1,
      col2: 2,
    };

    await container.items.upsert(item);
  } catch (err) {
    console.log(err.message);
  }
}

upsert();

挙動

競合するデータが存在しない場合は、INSERT、
競合するデータが存在する場合は、UPDATE
が基本的な UPSERT の挙動。

データの一意性、競合については、前回記事参照。

1．競合するデータが存在しない場合

UPSERT すると、以下のようにアイテムが追加される。

2．Partition Key が競合するデータが存在する場合

pk、id が一致するデータの UPSERT

{"pk": "A", "id": "0001", "col1": 1, "col2": 2} が存在する状態で、
{"pk": "A", "id": "0001", "col1": 2, "col2": 3} を UPSERT すると、
以下のようにアイテムが追加されることなく col1、col2 の値が UPDATE される。

3．Unique keys が競合するデータが存在する場合

pk と、col1、col2 が一致するデータの UPSERT

{"pk": "A", "id": "0001", "col1": 1, "col2": 2} が存在する状態で、
{"pk": "A", "id": "0002", "col1": 1, "col2": 2, "col3": 3} を UPSERT すると、
競合が発生する。

つまり、Unique keys の競合の場合は UPDATE されない。
そのため、Unique keys の競合に対しては create でも挙動は同じ。

競合例外発生時に UPDATE を実行すれば、UPSERT は実現可能。

一括 UPSERT

使用コード

一括 UPSERT には SDK の bulk を使用する。

const { CosmosClient } = require('@azure/cosmos');
const config = require('./config');

async function bulkUpsert() {
  const { connectionString, databaseId, containerId } = config;

  const client = new CosmosClient(connectionString);
  const database = client.database(databaseId);
  const container = database.container(containerId);

  const operations = [
    {
      // INSERT
      operationType: 'Upsert',
      partitionKey: 'A',
      resourceBody: {
        pk: 'A',
        id: '0001',
        col1: 1,
        col2: 2,
      },
    },
    {
      // col1, col2 が同じため CONFLICT
      operationType: 'Upsert',
      partitionKey: 'A',
      resourceBody: {
        pk: 'A',
        id: '0002',
        col1: 1,
        col2: 2,
      },
    },
    {
      // col1, col2 が同じため CONFLICT
      operationType: 'Upsert',
      partitionKey: 'A',
      resourceBody: {
        pk: 'A',
        id: '0003',
        col1: 1,
        col2: 2,
        col3: 3,
      },
    },
    {
      // INSERT
      operationType: 'Upsert',
      partitionKey: 'A',
      resourceBody: {
        pk: 'A',
        id: '0004',
        col1: 1,
        col2: 3,
      },
    },
  ];
  bulkOptions = { continueOnError: true };

  try {
    await container.items.bulk(operations, bulkOptions);
  } catch (err) {
    console.log(err.message);
  }
}

bulkUpsert();

挙動

単発での UPSERT と同様。
ただし、競合が発生したアイテムに対して個別に例外処理を実行できないため、
UPSERT は実現不可。

上記コードのように
オプションで { continueOnError: true } を指定することで例外を無視できるため、

競合するデータが存在しない場合は、INSERT、
競合するデータが存在する場合は、IGNORE

は実現可能。

上記コードを実行した場合、
id: '0001' は INSERT、
id: '0002' は col1, col2 が同じため CONFLICT
id: '0003' は col1, col2 が同じため CONFLICT
id: '0004' は INSERT
となる。