【PL/SQL】カーソルFORループをBULK化する判断基準｜実測と安全な分割処理

BEGIN
  -- 行ごとのIFとUPDATEを、条件付きUPDATEへ集約する
  UPDATE app.orders o
     SET o.status = 'ON_HOLD',
         o.updated_at = SYSTIMESTAMP
   WHERE o.status = 'PENDING'
     AND EXISTS (
           SELECT 1
             FROM app.customers c
            WHERE c.customer_id = o.customer_id
              AND o.amount > c.credit_limit
         );

  DBMS_OUTPUT.PUT_LINE('updated=' || SQL%ROWCOUNT);
  COMMIT;
END;
/

SET SERVEROUTPUT ON

DECLARE
  TYPE t_order_tab IS TABLE OF app.orders%ROWTYPE;
  v_rows       t_order_tab;
  v_started    PLS_INTEGER;
  v_checksum   NUMBER := 0;

  CURSOR c_orders IS
    SELECT *
      FROM app.orders
     WHERE status = 'PENDING'
     ORDER BY order_id;
BEGIN
  v_started := DBMS_UTILITY.GET_TIME;
  FOR r IN c_orders LOOP
    v_checksum := v_checksum + NVL(r.amount, 0);
  END LOOP;
  DBMS_OUTPUT.PUT_LINE(
    'cursor_for_loop_cs=' || (DBMS_UTILITY.GET_TIME - v_started)
  );

  v_checksum := 0;
  v_started := DBMS_UTILITY.GET_TIME;
  OPEN c_orders;
  LOOP
    FETCH c_orders BULK COLLECT INTO v_rows LIMIT 500;
    EXIT WHEN v_rows.COUNT = 0;

    FOR i IN 1 .. v_rows.COUNT LOOP
      v_checksum := v_checksum + NVL(v_rows(i).amount, 0);
    END LOOP;
  END LOOP;
  CLOSE c_orders;

  DBMS_OUTPUT.PUT_LINE(
    'bulk_limit_cs=' || (DBMS_UTILITY.GET_TIME - v_started)
  );
END;
/

DECLARE
  TYPE t_order_tab IS TABLE OF app.orders%ROWTYPE;
  v_rows      t_order_tab;
  v_limit     CONSTANT PLS_INTEGER := 500;

  CURSOR c_target IS
    SELECT *
      FROM app.orders
     WHERE status = 'PENDING'
       AND amount >= 1000
     ORDER BY order_id;
BEGIN
  OPEN c_target;
  LOOP
    FETCH c_target BULK COLLECT INTO v_rows LIMIT v_limit;
    EXIT WHEN v_rows.COUNT = 0;

    -- PL/SQLでしか表現しにくい検証を、この配列に対して実施する
    FOR i IN 1 .. v_rows.COUNT LOOP
      v_rows(i).status := 'REVIEWED';
    END LOOP;

    FORALL i IN 1 .. v_rows.COUNT
      UPDATE app.orders
         SET status = v_rows(i).status,
             updated_at = SYSTIMESTAMP
       WHERE order_id = v_rows(i).order_id;
  END LOOP;
  CLOSE c_target;

  COMMIT;
EXCEPTION
  WHEN OTHERS THEN
    IF c_target%ISOPEN THEN
      CLOSE c_target;
    END IF;
    ROLLBACK;
    RAISE;
END;
/

SELECT
  n.name,
  s.value
FROM v$sesstat s
JOIN v$statname n
  ON n.statistic# = s.statistic#
WHERE s.sid = SYS_CONTEXT('USERENV', 'SID')
  AND n.name IN (
    'session pga memory',
    'session pga memory max',
    'CPU used by this session',
    'session logical reads'
  )
ORDER BY n.name;

DECLARE
  TYPE t_id_tab IS TABLE OF app.orders.order_id%TYPE
    INDEX BY PLS_INTEGER;
  TYPE t_status_tab IS TABLE OF app.orders.status%TYPE
    INDEX BY PLS_INTEGER;

  v_ids       t_id_tab;
  v_statuses  t_status_tab;
  e_bulk_errors EXCEPTION;
  PRAGMA EXCEPTION_INIT(e_bulk_errors, -24381);
BEGIN
  v_ids(1) := 1001;
  v_statuses(1) := 'DONE';
  v_ids(2) := 1002;
  v_statuses(2) := 'DONE';

  BEGIN
    FORALL i IN INDICES OF v_ids SAVE EXCEPTIONS
      UPDATE app.orders
         SET status = v_statuses(i),
             updated_at = SYSTIMESTAMP
       WHERE order_id = v_ids(i);
  EXCEPTION
    WHEN e_bulk_errors THEN
      FOR j IN 1 .. SQL%BULK_EXCEPTIONS.COUNT LOOP
        INSERT INTO app.order_retry_queue(
          order_id,
          error_code,
          error_message,
          retry_status,
          created_at
        ) VALUES (
          v_ids(SQL%BULK_EXCEPTIONS(j).ERROR_INDEX),
          SQL%BULK_EXCEPTIONS(j).ERROR_CODE,
          SQLERRM(-SQL%BULK_EXCEPTIONS(j).ERROR_CODE),
          'WAITING',
          SYSTIMESTAMP
        );
      END LOOP;
  END;

  COMMIT;
EXCEPTION
  WHEN OTHERS THEN
    ROLLBACK;
    RAISE;
END;
/

DECLARE
  v_run_id       NUMBER := 2026060601;
  v_last_id      app.batch_run_checkpoint.last_order_id%TYPE;
  v_next_last_id app.orders.order_id%TYPE;
BEGIN
  SELECT last_order_id
    INTO v_last_id
    FROM app.batch_run_checkpoint
   WHERE run_id = v_run_id
   FOR UPDATE;

  SELECT MAX(order_id)
    INTO v_next_last_id
    FROM (
      SELECT order_id
        FROM app.orders
       WHERE order_id > v_last_id
         AND status = 'READY'
       ORDER BY order_id
       FETCH FIRST 500 ROWS ONLY
    );

  IF v_next_last_id IS NOT NULL THEN
    UPDATE app.orders
       SET status = 'DONE',
           updated_at = SYSTIMESTAMP
     WHERE order_id > v_last_id
       AND order_id <= v_next_last_id
       AND status = 'READY';

    UPDATE app.batch_run_checkpoint
       SET last_order_id = v_next_last_id,
           updated_at = SYSTIMESTAMP
     WHERE run_id = v_run_id;

    COMMIT;
  ELSE
    -- 対象がなくても、チェックポイント行のロックを解放する
    COMMIT;
  END IF;
EXCEPTION
  WHEN OTHERS THEN
    ROLLBACK;
    RAISE;
END;
/

DECLARE
  TYPE t_job_id_tab IS TABLE OF app.jobs.job_id%TYPE;
  v_job_ids    t_job_id_tab;
  v_worker_id  VARCHAR2(64) := SYS_CONTEXT('USERENV', 'SESSION_USER')
                              || ':' || SYS_CONTEXT('USERENV', 'SID');
  v_limit      CONSTANT PLS_INTEGER := 200;
  v_cursor_open BOOLEAN := FALSE;

  CURSOR c_claim IS
    SELECT job_id
      FROM app.jobs
     WHERE status = 'READY'
       AND ROWNUM <= v_limit
     FOR UPDATE SKIP LOCKED;
BEGIN
  OPEN c_claim;
  v_cursor_open := TRUE;
  FETCH c_claim BULK COLLECT INTO v_job_ids LIMIT v_limit;
  CLOSE c_claim;
  v_cursor_open := FALSE;

  IF v_job_ids.COUNT = 0 THEN
    COMMIT;
    RETURN;
  END IF;

  FORALL i IN 1 .. v_job_ids.COUNT
    UPDATE app.jobs
       SET status = 'CLAIMED',
           claimed_by = v_worker_id,
           claimed_at = SYSTIMESTAMP
     WHERE job_id = v_job_ids(i);

  COMMIT;

  -- 重い実処理はロック解放後に行う。
  -- 完了時は job_id と claimed_by の両方を条件にしてDONEへ更新する。
EXCEPTION
  WHEN OTHERS THEN
    IF v_cursor_open THEN
      CLOSE c_claim;
    END IF;
    ROLLBACK;
    RAISE;
END;
/

CREATE TYPE order_amount_obj AS OBJECT(
  order_id NUMBER,
  amount   NUMBER
);
/
CREATE TYPE order_amount_tab AS TABLE OF order_amount_obj;
/

CREATE OR REPLACE FUNCTION stream_large_orders
  RETURN order_amount_tab PIPELINED
IS
BEGIN
  FOR r IN (
    SELECT order_id, amount
      FROM app.orders
     WHERE status = 'NEW'
       AND amount >= 1000
     ORDER BY order_id
  ) LOOP
    PIPE ROW(order_amount_obj(r.order_id, r.amount));
  END LOOP;
  RETURN;
END;
/