PostgreSQLのUPSERTエラー「cannot affect row a second time」の解決方法

問題の背景

大規模なデータセット（50MBのCSVファイルや外部APIからのJSONペイロードなど）をターゲットテーブルに同期しようとしているとします。コード自体は問題なさそうに見えても、一括（bulk）で UPSERT を実行した瞬間に、PostgreSQLがトランザクションを停止させることがあります。これは、入力データ内に同じユニークキー（Primary Keyやメールアドレスなど）を持つエントリが複数含まれている場合に発生します。

データベースエンジンは予測可能性を求めます。1つのコマンドで同じ物理的な行を2回更新しようとした場合、どの値のセットを「優先」すべきかをエンジンが勝手に推測することはありません。データの整合性を維持するため、PostgreSQLは恣意的な選択をするのではなく、プロセス全体を停止させます。

エラーメッセージ

ERROR: ON CONFLICT DO UPDATE command cannot affect row a second time
HINT: Ensure that no rows proposed for insertion within the same command have duplicate constrained values.

発生原因

根本的な原因は、ソースデータ内に重複するキーが存在することです。PostgreSQLは INSERT ステートメントを単一のアトミックな操作として処理します。バッチ内に同じユニークIDを持つ行が2つある場合、エンジンは次のようなロジックループに陥ります。

1行目を挿入しようとする（または既存の場合は更新する）。
同じバッチ内の2つ目のインスタンスに到達した際、その同じ行を再度更新しようとする。

PostgreSQLはこれを明示的に禁止しています。最初の値、最後の値、あるいは両方の組み合わせのどれを求めているのか判断できないからです。データの衝突リスクを冒す代わりに、エラーをスローします。

具体的な例

イメージとして、sku にユニーク制約がある product_inventory というEコマース用のテーブルを想定してください。次のようなバッチ更新を受け取ったとします。

INSERT INTO product_inventory (sku, stock_count)
VALUES 
  ('LAPTOP-001', 50), 
  ('PHONE-002', 120), 
  ('LAPTOP-001', 45) -- 同じバッチ内に重複するSKUが存在する！
ON CONFLICT (sku) 
DO UPDATE SET stock_count = EXCLUDED.stock_count;

LAPTOP-001 が2回出現するため、データベースは最終的な在庫数を50にすべきか45にすべきか判断できません。その結果、トランザクションは即座に失敗します。

ステップバイステップの解決策

これを解決するには、データが INSERT ステートメントに到達する前にクレンジングを行う必要があります。ここでは、信頼性の高い3つの対処法を紹介します。

方法1：DISTINCT ONを使用する（高速かつシンプル）

重複の中から1つのレコードだけを選択して処理を進めたい場合に最適です。サブクエリ内で DISTINCT ON を使用して入力をフィルタリングします。

INSERT INTO product_inventory (sku, stock_count)
SELECT DISTINCT ON (sku) sku, stock_count
FROM (
    VALUES 
      ('LAPTOP-001', 50), 
      ('PHONE-002', 120), 
      ('LAPTOP-001', 45)
) AS input_data(sku, stock_count)
ORDER BY sku, stock_count DESC -- 重複がある場合は最大の在庫数を選択する
ON CONFLICT (sku) 
DO UPDATE SET stock_count = EXCLUDED.stock_count;

方法2：最新レコードを特定するためにCTEを使用する

データにタイムスタンプが含まれている場合、通常は最新の更新内容を保持したいはずです。1,000行を超えるようなバッチ処理では、ROW_NUMBER() を使用した共通テーブル式（CTE）を利用するのが最もプロフェッショナルな方法です。

WITH cleaned_updates AS (
    SELECT 
        sku, 
        stock_count, 
        updated_at,
        ROW_NUMBER() OVER (PARTITION BY sku ORDER BY updated_at DESC) as rank
    FROM (
        VALUES 
          ('LAPTOP-001', 50, '2023-10-25 10:00:00'::timestamp), 
          ('PHONE-002', 120, '2023-10-25 10:05:00'::timestamp), 
          ('LAPTOP-001', 45, '2023-10-25 10:10:00'::timestamp)
    ) AS raw_data(sku, stock_count, updated_at)
)
INSERT INTO product_inventory (sku, stock_count)
SELECT sku, stock_count
FROM cleaned_updates
WHERE rank = 1
ON CONFLICT (sku) 
DO UPDATE SET stock_count = EXCLUDED.stock_count;

方法3：値を事前に集約する

財務データやヒットカウンターなど、値を上書きするのではなく合計する必要がある場合は、まずデータを集約します。

INSERT INTO daily_sales (product_id, total_sold)
SELECT product_id, SUM(amount)
FROM (VALUES (101, 1), (101, 2), (102, 5)) AS batch(product_id, amount)
GROUP BY product_id
ON CONFLICT (product_id) 
DO UPDATE SET total_sold = daily_sales.total_sold + EXCLUDED.total_sold;

修正の検証方法

クエリを実行して結果を祈るだけではいけません。データがクリーンであることを確認するために、次の3つのステップを実行してください。

SELECT文を分離する: リファクタリングしたコードのサブクエリまたはCTEの部分だけを実行します。行数が期待されるユニークキーの数と一致することを確認してください。
データ損失のチェック: 1,000行から開始して最終的に800行になった場合、「消えた」200行が確かに重複であったことを確認します。
エッジケースのテスト: 3つの同一キーを含む小さなバッチを試してください。ORDER BY のロジックが正しければ、データベースは実行のたびに一貫して同じレコードを選択するはずです。

重要なポイント

1つのステートメントにつき、1つの行: PostgreSQLは単一のコマンドを1つの作業単位として見なします。整合性を損なうことなく、その単位内で同じ行を2回変更することはできません。
ソースデータのサニタイズ: 外部データを決して信用しないでください。CSVやAPIのペイロードには重複が含まれていると常に想定し、UPSERT を実行する前にデデュープ（重複排除）レイヤーを挟むようにしましょう。
明示的に指定する: 重複を排除する際は ORDER BY を使用してください。これにより UPSERT の動作が予測可能になり、実行ごとにランダムに結果が変わることを防げます。
ターゲットを一致させる: この修正を機能させるには、ON CONFLICT 句に指定する列と、DISTINCT ON または PARTITION BY 句に指定する列を一致させる必要があります。