【Oracle】DBMS_PARALLEL_EXECUTE完全ガイド｜大規模テーブルをチャンク分割して並列バッチ更新・エラー再実行まで解説

-- ステップ1: タスクを作成する
EXEC DBMS_PARALLEL_EXECUTE.CREATE_TASK(task_name => 'update_status_task');

-- ステップ2: テーブルを ROWID 範囲でチャンク分割する
-- chunk_size: 1チャンクあたりの ROWID 数（目安: 10000〜100000）
EXEC DBMS_PARALLEL_EXECUTE.CREATE_CHUNKS_BY_ROWID(
    task_name   => 'update_status_task',
    table_owner => 'HR',
    table_name  => 'ORDERS',
    by_row      => TRUE,        -- TRUE: 行数単位, FALSE: ブロック数単位
    chunk_size  => 10000        -- 1チャンクあたり 10,000 行
);

-- チャンクが作成されたか確認する
SELECT chunk_id, status, start_rowid, end_rowid
FROM   USER_PARALLEL_EXECUTE_CHUNKS
WHERE  task_name = 'update_status_task'
ORDER  BY chunk_id;

-- ステップ3: 各チャンクを処理する SQL を定義して実行する
-- SQL には :start_id（開始 ROWID）と :end_id（終了 ROWID）のプレースホルダを使う
DECLARE
    v_sql VARCHAR2(1000);
BEGIN
    v_sql := 'UPDATE orders
              SET    status = ''ARCHIVED''
              WHERE  order_date < DATE ''2023-01-01''
                AND  ROWID BETWEEN :start_id AND :end_id';

    DBMS_PARALLEL_EXECUTE.RUN_TASK(
        task_name      => 'update_status_task',
        sql_stmt       => v_sql,
        language_flag  => DBMS_SQL.NATIVE,
        parallel_level => 4        -- 並列度（同時実行するジョブ数）
    );
END;
/

-- ステップ4: タスクのステータスを確認する
SELECT status FROM USER_PARALLEL_EXECUTE_TASKS WHERE task_name = 'update_status_task';
-- FINISHED: すべてのチャンクが成功
-- FINISHED_WITH_ERROR: 一部チャンクが失敗

-- ステップ5: 完了後にタスクを削除する
EXEC DBMS_PARALLEL_EXECUTE.DROP_TASK(task_name => 'update_status_task');

-- CREATE_TASK（同様に実行）
EXEC DBMS_PARALLEL_EXECUTE.CREATE_TASK(task_name => 'update_orders_by_id');

-- 数値列（order_id）でチャンク分割する
EXEC DBMS_PARALLEL_EXECUTE.CREATE_CHUNKS_BY_NUMBER_COL(
    task_name   => 'update_orders_by_id',
    table_owner => 'HR',
    table_name  => 'ORDERS',
    table_column => 'ORDER_ID',    -- 分割に使う数値列（主キーや連番が適切）
    chunk_size  => 50000            -- 1チャンクあたり 50,000 件
);

-- チャンクの内容を確認する
SELECT chunk_id, status,
       start_id AS start_order_id,  -- 開始 ORDER_ID
       end_id   AS end_order_id     -- 終了 ORDER_ID
FROM   USER_PARALLEL_EXECUTE_CHUNKS
WHERE  task_name = 'update_orders_by_id'
ORDER  BY chunk_id;

-- NUMBER 列チャンクでは :start_id / :end_id が数値になる
DECLARE
    v_sql VARCHAR2(1000);
BEGIN
    v_sql := 'UPDATE orders
              SET    discount_applied = ''Y''
              WHERE  order_id BETWEEN :start_id AND :end_id
                AND  total_amount > 100000';

    DBMS_PARALLEL_EXECUTE.RUN_TASK(
        task_name      => 'update_orders_by_id',
        sql_stmt       => v_sql,
        language_flag  => DBMS_SQL.NATIVE,
        parallel_level => 8        -- より多くの並列ジョブで処理する
    );
END;
/

-- 失敗したチャンクを確認する
SELECT chunk_id, status, error_code, error_message,
       start_id, end_id
FROM   USER_PARALLEL_EXECUTE_CHUNKS
WHERE  task_name = 'update_orders_by_id'
  AND  status = 'PROCESSED_WITH_ERROR'
ORDER  BY chunk_id;
-- status の値:
-- UNASSIGNED        : 未処理（初期状態）
-- ASSIGNED          : 処理中（ジョブに割り当て済み）
-- PROCESSED         : 処理完了（成功）
-- PROCESSED_WITH_ERROR: 処理失敗（error_message でエラー内容を確認する）

-- 失敗チャンクだけを再実行する（RESUME_TASK）
-- 成功済みチャンクはスキップされる
DECLARE
    v_sql VARCHAR2(1000);
BEGIN
    v_sql := 'UPDATE orders
              SET    discount_applied = ''Y''
              WHERE  order_id BETWEEN :start_id AND :end_id
                AND  total_amount > 100000';

    -- RUN_TASK の代わりに RESUME_TASK を使う（失敗チャンクのみ再実行）
    DBMS_PARALLEL_EXECUTE.RESUME_TASK(
        task_name      => 'update_orders_by_id',
        sql_stmt       => v_sql,
        language_flag  => DBMS_SQL.NATIVE,
        parallel_level => 4
    );
END;
/

-- 再実行後に全チャンクが成功したか確認する
SELECT status, COUNT(*) AS chunk_count
FROM   USER_PARALLEL_EXECUTE_CHUNKS
WHERE  task_name = 'update_orders_by_id'
GROUP  BY status;
-- status = PROCESSED だけになれば完了

-- タスク一覧と全体ステータスを確認する
SELECT task_name, status, start_ts, end_ts,
       job_prefix,    -- DBMS_SCHEDULER ジョブのプレフィックス
       chunk_size,
       chunk_type     -- ROWID or NUMBER_RANGE
FROM   USER_PARALLEL_EXECUTE_TASKS
ORDER  BY start_ts DESC;

-- 進捗をリアルタイムに確認する
SELECT
    COUNT(*) AS total_chunks,
    SUM(CASE status WHEN 'PROCESSED'             THEN 1 ELSE 0 END) AS done,
    SUM(CASE status WHEN 'PROCESSED_WITH_ERROR'  THEN 1 ELSE 0 END) AS errors,
    SUM(CASE status WHEN 'ASSIGNED'              THEN 1 ELSE 0 END) AS running,
    SUM(CASE status WHEN 'UNASSIGNED'            THEN 1 ELSE 0 END) AS pending
FROM   USER_PARALLEL_EXECUTE_CHUNKS
WHERE  task_name = 'update_orders_by_id';

-- 実行中の DBMS_SCHEDULER ジョブを確認する
SELECT job_name, state, last_start_date
FROM   USER_SCHEDULER_RUNNING_JOBS
WHERE  job_name LIKE 'CHUNKED_JOB%'  -- RUN_TASK が作成するジョブ名のパターン
ORDER  BY last_start_date;

-- 注意: parallel_level 個の DBMS_SCHEDULER ジョブが作成される
-- ジョブはチャンクを順番に取りに行きながら処理するキューイング方式