【PL/SQL】差分抽出・増分処理の設計｜前回実行日時・再実行・重複防止まで

CREATE TABLE incremental_job_state (
  job_name              VARCHAR2(100) PRIMARY KEY,
  last_success_to       TIMESTAMP,
  overlap_minutes       NUMBER DEFAULT 5 NOT NULL,
  status                VARCHAR2(20) DEFAULT 'IDLE' NOT NULL,
  last_started_at       TIMESTAMP,
  last_finished_at      TIMESTAMP,
  last_error_message    VARCHAR2(2000),
  updated_by            VARCHAR2(128) DEFAULT SYS_CONTEXT('USERENV','SESSION_USER') NOT NULL,
  updated_at            TIMESTAMP DEFAULT SYSTIMESTAMP NOT NULL,
  CONSTRAINT ck_incremental_state_status
    CHECK (status IN ('IDLE','RUNNING','FAILED'))
);

INSERT INTO incremental_job_state (
  job_name,
  last_success_to,
  overlap_minutes
) VALUES (
  'ORDER_INCREMENTAL_LOAD',
  TIMESTAMP '2026-06-01 00:00:00',
  5
);

DECLARE
  v_job_name  VARCHAR2(100) := 'ORDER_INCREMENTAL_LOAD';
  v_from_ts   TIMESTAMP;
  v_to_ts     TIMESTAMP;
  v_overlap   NUMBER;
BEGIN
  SELECT
    last_success_to,
    SYSTIMESTAMP,
    overlap_minutes
  INTO
    v_from_ts,
    v_to_ts,
    v_overlap
  FROM incremental_job_state
  WHERE job_name = v_job_name
  FOR UPDATE;

  v_from_ts := v_from_ts - NUMTODSINTERVAL(v_overlap, 'MINUTE');

  DBMS_APPLICATION_INFO.SET_MODULE(
    module_name => v_job_name,
    action_name => 'from=' || TO_CHAR(v_from_ts, 'YYYY-MM-DD HH24:MI:SS')
  );

  -- ここから v_from_ts 以上、v_to_ts 未満を処理する
END;
/

CREATE TABLE stg_order_incremental (
  batch_id        NUMBER       NOT NULL,
  source_order_id NUMBER       NOT NULL,
  customer_id     NUMBER,
  order_status    VARCHAR2(30),
  amount          NUMBER,
  source_updated_at TIMESTAMP,
  row_hash        VARCHAR2(64),
  process_status  VARCHAR2(20) DEFAULT 'NEW' NOT NULL,
  error_message    VARCHAR2(2000),
  loaded_at        TIMESTAMP DEFAULT SYSTIMESTAMP NOT NULL,
  CONSTRAINT pk_stg_order_incremental
    PRIMARY KEY (batch_id, source_order_id)
);

INSERT INTO stg_order_incremental (
  batch_id,
  source_order_id,
  customer_id,
  order_status,
  amount,
  source_updated_at,
  row_hash
)
SELECT
  :batch_id,
  o.order_id,
  o.customer_id,
  o.order_status,
  o.amount,
  o.updated_at,
  STANDARD_HASH(
    NVL(TO_CHAR(o.customer_id), '#NULL#') || '|'
    || NVL(o.order_status, '#NULL#') || '|'
    || NVL(TO_CHAR(o.amount, 'FM9999999999990D00', 'NLS_NUMERIC_CHARACTERS=.,'), '#NULL#'),
    'SHA256'
  ) AS row_hash
FROM source_orders o
WHERE o.updated_at >= :from_ts
  AND o.updated_at <  :to_ts;

MERGE INTO order_summary d
USING (
  SELECT
    source_order_id,
    customer_id,
    order_status,
    amount,
    source_updated_at,
    row_hash
  FROM stg_order_incremental
  WHERE batch_id = :batch_id
    AND process_status = 'NEW'
) s
ON (d.source_order_id = s.source_order_id)
WHEN MATCHED THEN
  UPDATE SET
    d.customer_id = s.customer_id,
    d.order_status = s.order_status,
    d.amount = s.amount,
    d.source_updated_at = s.source_updated_at,
    d.row_hash = s.row_hash,
    d.updated_at = SYSTIMESTAMP
  WHERE NVL(d.row_hash, '-') <> NVL(s.row_hash, '-')
WHEN NOT MATCHED THEN
  INSERT (
    source_order_id,
    customer_id,
    order_status,
    amount,
    source_updated_at,
    row_hash,
    created_at,
    updated_at
  )
  VALUES (
    s.source_order_id,
    s.customer_id,
    s.order_status,
    s.amount,
    s.source_updated_at,
    s.row_hash,
    SYSTIMESTAMP,
    SYSTIMESTAMP
  );

UPDATE incremental_job_state
SET
  last_success_to = :to_ts,
  status = 'IDLE',
  last_finished_at = SYSTIMESTAMP,
  last_error_message = NULL,
  updated_at = SYSTIMESTAMP
WHERE job_name = 'ORDER_INCREMENTAL_LOAD';

COMMIT;

-- まず本処理の変更を戻す
ROLLBACK;

-- その後、失敗状態を別トランザクションとして記録する
UPDATE incremental_job_state
SET
  status = 'FAILED',
  last_finished_at = SYSTIMESTAMP,
  last_error_message = :error_message,
  updated_at = SYSTIMESTAMP
WHERE job_name = 'ORDER_INCREMENTAL_LOAD';

COMMIT;

WHERE (
      o.updated_at >= :from_ts
  AND o.updated_at <  :to_ts
)
OR (
      o.received_at >= :from_ts
  AND o.received_at <  :to_ts
)

SELECT last_success_to, overlap_minutes
INTO   v_last_success_to, v_overlap_minutes
FROM   incremental_job_state
WHERE  job_name = 'ORDER_INCREMENTAL_LOAD'
FOR UPDATE NOWAIT;

UPDATE incremental_job_state
SET
  status = 'RUNNING',
  last_started_at = SYSTIMESTAMP,
  updated_at = SYSTIMESTAMP
WHERE job_name = 'ORDER_INCREMENTAL_LOAD';

SELECT
  source_order_id,
  source_updated_at,
  'ORDER:' || source_order_id || ':'
    || TO_CHAR(source_updated_at, 'YYYYMMDDHH24MISSFF3') AS idempotency_key
FROM stg_order_incremental
WHERE batch_id = :batch_id;

SELECT
  TRUNC(updated_at, 'HH24') AS chunk_hour,
  COUNT(*) AS rows_count
FROM source_orders
WHERE updated_at >= :from_ts
  AND updated_at <  :to_ts
GROUP BY TRUNC(updated_at, 'HH24')
ORDER BY chunk_hour;