【PL/SQL】MERGE文でUPSERTを高速・安全に実装｜IF方式8倍差・LOG ERRORS・ORA-30926回避・実務10シナリオ

BEGIN
  MERGE INTO products t
  USING (
    SELECT :p_code AS code,
           :p_name AS name,
           :p_price AS price
    FROM dual
  ) s
  ON (t.product_code = s.code)
  WHEN MATCHED THEN
    UPDATE SET
      t.product_name = s.name,
      t.price        = s.price,
      t.updated_at   = SYSDATE
  WHEN NOT MATCHED THEN
    INSERT (product_code, product_name, price, created_at, updated_at)
    VALUES (s.code, s.name, s.price, SYSDATE, SYSDATE);

  DBMS_OUTPUT.PUT_LINE('処理件数: ' || SQL%ROWCOUNT);
  COMMIT;
END;
/

-- ❌ レガシー実装：IF方式
DECLARE
  v_cnt NUMBER;
BEGIN
  FOR rec IN (SELECT * FROM import_products) LOOP
    SELECT COUNT(*) INTO v_cnt
    FROM products WHERE product_code = rec.code;

    IF v_cnt > 0 THEN
      UPDATE products
         SET product_name = rec.name,
             price = rec.price,
             updated_at = SYSDATE
       WHERE product_code = rec.code;
    ELSE
      INSERT INTO products(product_code, product_name, price, created_at, updated_at)
      VALUES(rec.code, rec.name, rec.price, SYSDATE, SYSDATE);
    END IF;
  END LOOP;
  COMMIT;
END;
/

-- ✅ MERGE実装：1文で済む
BEGIN
  MERGE INTO products t
  USING import_products s
  ON (t.product_code = s.code)
  WHEN MATCHED THEN
    UPDATE SET t.product_name = s.name,
               t.price = s.price,
               t.updated_at = SYSDATE
  WHEN NOT MATCHED THEN
    INSERT(product_code, product_name, price, created_at, updated_at)
    VALUES(s.code, s.name, s.price, SYSDATE, SYSDATE);
  COMMIT;
END;
/

DECLARE
  v_total NUMBER;
BEGIN
  MERGE INTO products t
  USING import_products s
  ON (t.product_code = s.code)
  WHEN MATCHED THEN
    UPDATE SET t.price = s.price, t.updated_at = SYSDATE
  WHEN NOT MATCHED THEN
    INSERT(product_code, price, created_at)
    VALUES(s.code, s.price, SYSDATE);

  v_total := SQL%ROWCOUNT;  -- MERGE直後にだけ有効
  DBMS_OUTPUT.PUT_LINE('処理件数合計: ' || v_total);

  -- 後続のSQLを実行すると SQL%ROWCOUNT は上書きされるので注意
  COMMIT;
END;
/

-- 事前に sql_ops_tmp テーブルを用意（op CHAR(1) と key のみ）
BEGIN
  DELETE FROM sql_ops_tmp;

  MERGE INTO products t
  USING import_products s
  ON (t.product_code = s.code)
  WHEN MATCHED THEN
    UPDATE SET t.price = s.price, t.updated_at = SYSDATE
  WHEN NOT MATCHED THEN
    INSERT(product_code, price, created_at)
    VALUES(s.code, s.price, SYSDATE)
    -- INSERTされた行だけに印をつける別トリガーを使うか、
    -- 下記のように「存在確認→記録」を事前に行う方法が確実
  ;
END;
/

-- より確実な方法：MERGE直前にキーの存在状況を記録する
DECLARE
  v_match_cnt    NUMBER;
  v_total_cnt    NUMBER;
  v_updated_cnt  NUMBER;
  v_inserted_cnt NUMBER;
BEGIN
  -- 1. 事前にマッチ件数を数える
  SELECT COUNT(*)
    INTO v_match_cnt
    FROM products t
   WHERE EXISTS (
           SELECT 1 FROM import_products s
            WHERE s.code = t.product_code);

  -- 2. MERGE実行
  MERGE INTO products t
  USING import_products s
  ON (t.product_code = s.code)
  WHEN MATCHED THEN
    UPDATE SET t.price = s.price, t.updated_at = SYSDATE
  WHEN NOT MATCHED THEN
    INSERT(product_code, price, created_at)
    VALUES(s.code, s.price, SYSDATE);

  v_total_cnt    := SQL%ROWCOUNT;
  v_updated_cnt  := v_match_cnt;
  v_inserted_cnt := v_total_cnt - v_match_cnt;

  DBMS_OUTPUT.PUT_LINE('UPDATE件数: ' || v_updated_cnt);
  DBMS_OUTPUT.PUT_LINE('INSERT件数: ' || v_inserted_cnt);
  COMMIT;
END;
/

-- ① エラー記録表を作成（1回だけ）
BEGIN
  DBMS_ERRLOG.CREATE_ERROR_LOG(
    dml_table_name      => 'PRODUCTS',
    err_log_table_name  => 'ERR$_PRODUCTS'
  );
END;
/

-- ② MERGE に LOG ERRORS INTO を付加
BEGIN
  MERGE INTO products t
  USING import_products s
  ON (t.product_code = s.code)
  WHEN MATCHED THEN
    UPDATE SET t.price = s.price, t.updated_at = SYSDATE
  WHEN NOT MATCHED THEN
    INSERT(product_code, price, created_at)
    VALUES(s.code, s.price, SYSDATE)
  LOG ERRORS INTO err$_products ('20260423_BATCH')
  REJECT LIMIT UNLIMITED;  -- 何件エラーが出てもバッチは継続

  DBMS_OUTPUT.PUT_LINE('成功件数: ' || SQL%ROWCOUNT);
  COMMIT;
END;
/

-- ③ エラー行を調査
SELECT ora_err_number$, ora_err_mesg$, ora_err_tag$, product_code
  FROM err$_products
 WHERE ora_err_tag$ = '20260423_BATCH';

DECLARE
  TYPE t_codes IS TABLE OF products.product_code%TYPE;
  TYPE t_prices IS TABLE OF products.price%TYPE;
  v_codes  t_codes;
  v_prices t_prices;
  e_bulk_err EXCEPTION;
  PRAGMA EXCEPTION_INIT(e_bulk_err, -24381);
BEGIN
  SELECT code, price BULK COLLECT INTO v_codes, v_prices FROM import_products;

  BEGIN
    FORALL i IN 1..v_codes.COUNT SAVE EXCEPTIONS
      MERGE INTO products t
      USING (SELECT v_codes(i) AS code, v_prices(i) AS price FROM dual) s
      ON (t.product_code = s.code)
      WHEN MATCHED THEN
        UPDATE SET t.price = s.price
      WHEN NOT MATCHED THEN
        INSERT(product_code, price) VALUES(s.code, s.price);
  EXCEPTION
    WHEN e_bulk_err THEN
      FOR i IN 1..SQL%BULK_EXCEPTIONS.COUNT LOOP
        DBMS_OUTPUT.PUT_LINE(
          'NG行: ' || SQL%BULK_EXCEPTIONS(i).ERROR_INDEX ||
          ' / ' || SQLERRM(-SQL%BULK_EXCEPTIONS(i).ERROR_CODE));
      END LOOP;
  END;

  COMMIT;
END;
/

-- 事前準備（セッション単位）
ALTER SESSION ENABLE PARALLEL DML;

-- MERGEにヒントを付与
BEGIN
  MERGE /*+ APPEND PARALLEL(t, 4) */ INTO products t
  USING (
    SELECT /*+ PARALLEL(s, 4) */ code, price
      FROM import_products s
  ) s
  ON (t.product_code = s.code)
  WHEN MATCHED THEN
    UPDATE SET t.price = s.price, t.updated_at = SYSDATE
  WHEN NOT MATCHED THEN
    INSERT(product_code, price, created_at)
    VALUES(s.code, s.price, SYSDATE)
  LOG ERRORS INTO err$_products ('BATCH_' || TO_CHAR(SYSDATE,'YYYYMMDD'))
  REJECT LIMIT UNLIMITED;

  COMMIT;  -- 並列DML後は必ずCOMMITしてからセッションで他DMLを実行
END;
/

-- ❌ 発生例：import_productsに同じcodeが複数行ある
MERGE INTO products t
USING import_products s
ON (t.product_code = s.code)
WHEN MATCHED THEN UPDATE SET t.price = s.price
WHEN NOT MATCHED THEN INSERT(product_code, price) VALUES(s.code, s.price);
-- → ORA-30926: unable to get a stable set of rows in the source tables

-- ✅ 対策1: USINGでDISTINCT（重複行がすべて同一内容の場合）
MERGE INTO products t
USING (SELECT DISTINCT code, price FROM import_products) s
ON (t.product_code = s.code)
WHEN MATCHED THEN UPDATE SET t.price = s.price
WHEN NOT MATCHED THEN INSERT(product_code, price) VALUES(s.code, s.price);

-- ✅ 対策2: ROW_NUMBER()で1行だけ採用（重複行の内容が異なる場合）
MERGE INTO products t
USING (
  SELECT code, price FROM (
    SELECT code, price,
           ROW_NUMBER() OVER (PARTITION BY code ORDER BY updated_at DESC) rn
    FROM import_products
  ) WHERE rn = 1
) s
ON (t.product_code = s.code)
WHEN MATCHED THEN UPDATE SET t.price = s.price
WHEN NOT MATCHED THEN INSERT(product_code, price) VALUES(s.code, s.price);

MERGE INTO products t
USING import_products s
ON (t.product_code = s.code)
WHEN MATCHED THEN
  UPDATE SET
    t.product_name = s.name,
    t.price        = s.price,
    t.updated_at   = SYSDATE
  WHERE t.product_name != s.name                 -- どちらかが違うときだけ
     OR NVL(t.price, -1) != NVL(s.price, -1)    -- NULL対応
WHEN NOT MATCHED THEN
  INSERT(product_code, product_name, price, created_at, updated_at)
  VALUES(s.code, s.name, s.price, SYSDATE, SYSDATE);

-- シナリオ1: 日次差分ロード（数百万件、新規中心）
-- → APPEND + PARALLEL + LOG ERRORS、NOLOGGING表、夜間排他
MERGE /*+ APPEND PARALLEL(t,8) */ INTO fact_sales t
USING (SELECT /*+ PARALLEL(s,8) */ * FROM stg_sales_today s) s
ON (t.sales_id = s.sales_id)
WHEN NOT MATCHED THEN INSERT(...) VALUES(...)
LOG ERRORS INTO err$_fact_sales ('DAILY_'||TO_CHAR(SYSDATE,'YYYYMMDD'))
REJECT LIMIT 1000;

-- シナリオ2: 商品マスター同期（数千件、更新中心）
-- → 差分WHERE + ON索引活用、並列不要
MERGE INTO m_product t
USING (SELECT * FROM ext_product_csv) s
ON (t.product_code = s.code)
WHEN MATCHED THEN UPDATE SET t.name=s.name, t.price=s.price
  WHERE t.name!=s.name OR t.price!=s.price
WHEN NOT MATCHED THEN INSERT(product_code,name,price) VALUES(s.code,s.name,s.price);

-- シナリオ3: 在庫残高の加算集計
-- → USINGでSUM集計してからMERGE、ON条件で索引を使う
MERGE INTO stock t
USING (SELECT product_code, SUM(qty) qty FROM stock_movements
       WHERE move_date=TRUNC(SYSDATE) GROUP BY product_code) s
ON (t.product_code = s.product_code)
WHEN MATCHED THEN UPDATE SET t.qty = t.qty + s.qty
WHEN NOT MATCHED THEN INSERT(product_code,qty) VALUES(s.product_code,s.qty);

-- シナリオ4: 論理削除の同期（USINGに無い行を削除フラグON）
-- → MERGE本体とUPDATEの組み合わせ（MERGE単体では未マッチ側の削除は書けない）
MERGE INTO customers t
USING source_customers s
ON (t.customer_id = s.customer_id)
WHEN MATCHED THEN UPDATE SET t.is_deleted=0, t.name=s.name
WHEN NOT MATCHED THEN INSERT(customer_id,name,is_deleted) VALUES(s.customer_id,s.name,0);

UPDATE customers t SET t.is_deleted=1, t.updated_at=SYSDATE
 WHERE t.is_deleted=0
   AND NOT EXISTS(SELECT 1 FROM source_customers s WHERE s.customer_id=t.customer_id);

-- シナリオ5: 履歴化（更新のたびに履歴表へINSERT + マスターUPDATE）
-- → INSERTトリガー or MERGE + 別途 INSERT SELECT 方式
MERGE INTO m_price t
USING stg_price s
ON (t.product_code = s.code)
WHEN MATCHED THEN UPDATE SET t.price=s.price, t.updated_at=SYSDATE WHERE t.price!=s.price
WHEN NOT MATCHED THEN INSERT(product_code,price,updated_at) VALUES(s.code,s.price,SYSDATE);

INSERT INTO m_price_history(product_code, price, saved_at)
  SELECT product_code, price, SYSDATE FROM m_price
   WHERE updated_at = (SELECT MAX(updated_at) FROM m_price);

-- シナリオ6: 1件だけUPSERT（オンライン処理）
-- → USING DUAL パターンが最適
MERGE INTO sessions t
USING (SELECT :p_uid uid, SYSDATE last_at FROM dual) s
ON (t.user_id = s.uid)
WHEN MATCHED THEN UPDATE SET t.last_access_at = s.last_at
WHEN NOT MATCHED THEN INSERT(user_id, last_access_at) VALUES(s.uid, s.last_at);

-- シナリオ7: 外部ファイル（CSV）からのロード
-- → 外部表（EXTERNAL TABLE）+ MERGE
MERGE INTO m_employee t
USING (SELECT * FROM ext_employee_csv) s
ON (t.emp_no = s.emp_no)
WHEN MATCHED THEN UPDATE SET t.dept_code=s.dept, t.salary=s.salary
WHEN NOT MATCHED THEN INSERT(emp_no, dept_code, salary) VALUES(s.emp_no, s.dept, s.salary)
LOG ERRORS INTO err$_m_employee ('EMP_LOAD') REJECT LIMIT 100;

-- シナリオ8: 条件付きUPDATE（バージョンが新しい時だけ上書き）
-- → WHEN MATCHED AND ... で版比較
MERGE INTO documents t
USING stg_documents s
ON (t.doc_id = s.doc_id)
WHEN MATCHED AND s.version > t.version THEN
  UPDATE SET t.body=s.body, t.version=s.version
WHEN NOT MATCHED THEN INSERT(doc_id, body, version) VALUES(s.doc_id, s.body, s.version);

-- シナリオ9: 部分更新（一部カラムだけ更新したい）
-- → UPDATE句に必要な列だけ記述すればOK
MERGE INTO users t
USING stg_user_profile s
ON (t.user_id = s.uid)
WHEN MATCHED THEN UPDATE SET t.profile_json = s.profile_json, t.updated_at = SYSDATE
-- email や role は更新しない
WHEN NOT MATCHED THEN INSERT(user_id, profile_json) VALUES(s.uid, s.profile_json);

-- シナリオ10: 動的SQLでターゲット表を切り替え
-- → EXECUTE IMMEDIATE + バインド変数
DECLARE
  v_target VARCHAR2(30) := 'sales_2026_04';
  v_sql    CLOB;
BEGIN
  v_sql := 'MERGE INTO ' || v_target || ' t '
        || 'USING stg_sales s ON (t.sales_id=s.sales_id) '
        || 'WHEN MATCHED THEN UPDATE SET t.amount=s.amount '
        || 'WHEN NOT MATCHED THEN INSERT(sales_id,amount) VALUES(s.sales_id,s.amount)';
  EXECUTE IMMEDIATE v_sql;
  COMMIT;
END;
/