【PL/SQL】大量データ処理のコミット頻度とUNDO最適化｜チャンク設計・ORA-01555対策

DECLARE
  v_count PLS_INTEGER := 0;
BEGIN
  FOR r IN (
    SELECT order_id, customer_id
      FROM staging_orders
     WHERE processed = 'N'
     ORDER BY order_id
  ) LOOP
    process_order(r.order_id);
    v_count := v_count + 1;

    IF MOD(v_count, 1000) = 0 THEN
      -- NG: 業務単位を無視して機械的に区切っている
      COMMIT;
    END IF;
  END LOOP;

  COMMIT;
END;
/

DECLARE
  c_chunk_size CONSTANT NUMBER := 50000;
  v_from_id NUMBER;
  v_to_id   NUMBER;
BEGIN
  SELECT MIN(stg_id), MAX(stg_id)
    INTO v_from_id, v_to_id
    FROM staging_sales
   WHERE load_id = :load_id
     AND processed = 'N';

  WHILE v_from_id <= v_to_id LOOP
    UPDATE target_sales t
       SET amount = (
             SELECT s.amount
               FROM staging_sales s
              WHERE s.sale_id = t.sale_id
                AND s.stg_id BETWEEN v_from_id AND v_from_id + c_chunk_size - 1
           )
     WHERE EXISTS (
             SELECT 1
               FROM staging_sales s
              WHERE s.sale_id = t.sale_id
                AND s.stg_id BETWEEN v_from_id AND v_from_id + c_chunk_size - 1
           );

    INSERT INTO batch_progress(load_id, from_id, to_id, status, finished_at)
    VALUES (:load_id, v_from_id, v_from_id + c_chunk_size - 1, 'DONE', SYSTIMESTAMP);

    COMMIT;
    v_from_id := v_from_id + c_chunk_size;
  END LOOP;
EXCEPTION
  WHEN OTHERS THEN
    ROLLBACK;
    RAISE;
END;
/

DECLARE
  v_month DATE := DATE '2026-05-01';
BEGIN
  INSERT /*+ APPEND */ INTO sales_fact
  SELECT *
    FROM staging_sales
   WHERE sale_date >= v_month
     AND sale_date <  ADD_MONTHS(v_month, 1)
  LOG ERRORS INTO err$_sales_fact ('LOAD_202605')
  REJECT LIMIT UNLIMITED;

  COMMIT;
END;
/

MERGE INTO target_customer t
USING (
  SELECT customer_id, rank_code, updated_at
    FROM staging_customer
   WHERE load_id = :load_id
) s
ON (t.customer_id = s.customer_id)
WHEN MATCHED THEN
  UPDATE SET
    t.rank_code = s.rank_code,
    t.updated_at = s.updated_at
WHEN NOT MATCHED THEN
  INSERT (customer_id, rank_code, updated_at)
  VALUES (s.customer_id, s.rank_code, s.updated_at);

COMMIT;

BEGIN
  DBMS_ERRLOG.CREATE_ERROR_LOG(
    dml_table_name  => 'TARGET_SALES',
    err_log_table_name => 'ERR$_TARGET_SALES'
  );
END;
/

INSERT /*+ APPEND */ INTO target_sales(
  sale_id,
  sale_date,
  amount
)
SELECT sale_id,
       sale_date,
       amount
  FROM staging_sales
 WHERE load_id = :load_id
LOG ERRORS INTO err$_target_sales ('LOAD_202605')
REJECT LIMIT UNLIMITED;

COMMIT;

DECLARE
  CURSOR c_src IS
    SELECT id, value
      FROM staging_data
     WHERE load_id = :load_id
       AND processed = 'N'
     ORDER BY id;

  TYPE t_src IS TABLE OF c_src%ROWTYPE;
  l_rows t_src;
BEGIN
  OPEN c_src;
  LOOP
    FETCH c_src BULK COLLECT INTO l_rows LIMIT 10000;
    EXIT WHEN l_rows.COUNT = 0;

    BEGIN
      FORALL i IN 1 .. l_rows.COUNT SAVE EXCEPTIONS
        UPDATE target_data
           SET value = l_rows(i).value
         WHERE id = l_rows(i).id;
    EXCEPTION
      WHEN OTHERS THEN
        FOR i IN 1 .. SQL%BULK_EXCEPTIONS.COUNT LOOP
          INSERT INTO batch_error_log(error_index, error_code)
          VALUES (
            SQL%BULK_EXCEPTIONS(i).ERROR_INDEX,
            SQL%BULK_EXCEPTIONS(i).ERROR_CODE
          );
        END LOOP;
    END;

    COMMIT;
  END LOOP;
  CLOSE c_src;
END;
/

ALTER SESSION ENABLE PARALLEL DML;

INSERT /*+ APPEND PARALLEL(t 4) */ INTO target_sales t
SELECT /*+ PARALLEL(s 4) */
       *
  FROM staging_sales s
 WHERE s.load_id = :load_id;

COMMIT;

CREATE TABLE sales_new NOLOGGING AS
SELECT sale_id,
       sale_date,
       CASE
         WHEN status = 'OLD' THEN 'ARCHIVED'
         ELSE status
       END AS status,
       amount
  FROM sales_old;

-- 件数・合計・不整合を検証してから切り替える
SELECT COUNT(*), SUM(amount) FROM sales_old;
SELECT COUNT(*), SUM(amount) FROM sales_new;

-- 通常は明示指定せず、基盤設定と運用要件に従う
COMMIT;

-- 遅延許容の一部バッチで検討されることがある
COMMIT WRITE BATCH NOWAIT;

CREATE GLOBAL TEMPORARY TABLE work_target_keys (
  id NUMBER PRIMARY KEY
) ON COMMIT PRESERVE ROWS;

INSERT INTO work_target_keys(id)
SELECT id
  FROM staging_data
 WHERE load_id = :load_id;

COMMIT;

-- 以降は作業表のキーを範囲で処理する

CREATE TABLE batch_progress (
  batch_id      VARCHAR2(50),
  chunk_no      NUMBER,
  from_key      NUMBER,
  to_key        NUMBER,
  status        VARCHAR2(20),
  processed_cnt NUMBER,
  error_cnt     NUMBER,
  started_at    TIMESTAMP,
  finished_at   TIMESTAMP,
  error_message VARCHAR2(4000)
);