【PL/SQL】ステージングテーブル設計｜CSV取込・バリデーション・再実行・エラー隔離まで

CREATE TABLE import_batch (
  batch_id        NUMBER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
  source_name     VARCHAR2(255) NOT NULL,
  source_hash     VARCHAR2(64),
  status          VARCHAR2(20) DEFAULT 'STARTED' NOT NULL,
  total_count     NUMBER DEFAULT 0 NOT NULL,
  valid_count     NUMBER DEFAULT 0 NOT NULL,
  error_count     NUMBER DEFAULT 0 NOT NULL,
  loaded_count    NUMBER DEFAULT 0 NOT NULL,
  started_at      TIMESTAMP DEFAULT SYSTIMESTAMP NOT NULL,
  finished_at     TIMESTAMP,
  created_by      VARCHAR2(128) DEFAULT SYS_CONTEXT('USERENV','SESSION_USER') NOT NULL,
  last_error_msg  VARCHAR2(1000),
  CONSTRAINT ck_import_batch_status CHECK (
    status IN ('STARTED','LOADED_RAW','VALIDATED','LOADED','ERROR','CANCELLED')
  )
);

CREATE UNIQUE INDEX uq_import_batch_source_hash
  ON import_batch(source_hash);

CREATE TABLE stg_order_import (
  batch_id       NUMBER NOT NULL,
  row_no         NUMBER NOT NULL,
  raw_line       CLOB,
  order_no_raw   VARCHAR2(100),
  order_date_raw VARCHAR2(100),
  customer_raw   VARCHAR2(200),
  amount_raw     VARCHAR2(100),
  status         VARCHAR2(20) DEFAULT 'RAW' NOT NULL,
  error_code     VARCHAR2(100),
  error_msg      VARCHAR2(1000),
  loaded_order_id NUMBER,
  created_at     TIMESTAMP DEFAULT SYSTIMESTAMP NOT NULL,
  updated_at     TIMESTAMP DEFAULT SYSTIMESTAMP NOT NULL,
  CONSTRAINT pk_stg_order_import PRIMARY KEY (batch_id, row_no),
  CONSTRAINT ck_stg_order_status CHECK (
    status IN ('RAW','VALID','ERROR','LOADED','SKIPPED')
  ),
  CONSTRAINT fk_stg_order_batch FOREIGN KEY (batch_id)
    REFERENCES import_batch(batch_id)
);

CREATE INDEX ix_stg_order_status
  ON stg_order_import(batch_id, status);

CREATE OR REPLACE PACKAGE order_import_stage AS
  FUNCTION start_batch(
    p_source_name VARCHAR2,
    p_source_hash VARCHAR2
  ) RETURN NUMBER;

  PROCEDURE add_row(
    p_batch_id       NUMBER,
    p_row_no         NUMBER,
    p_raw_line       CLOB,
    p_order_no_raw   VARCHAR2,
    p_order_date_raw VARCHAR2,
    p_customer_raw   VARCHAR2,
    p_amount_raw     VARCHAR2
  );
END;
/

CREATE OR REPLACE PACKAGE BODY order_import_stage AS
  FUNCTION start_batch(
    p_source_name VARCHAR2,
    p_source_hash VARCHAR2
  ) RETURN NUMBER IS
    v_batch_id NUMBER;
  BEGIN
    INSERT INTO import_batch(source_name, source_hash)
    VALUES (p_source_name, p_source_hash)
    RETURNING batch_id INTO v_batch_id;

    RETURN v_batch_id;
  END;

  PROCEDURE add_row(
    p_batch_id       NUMBER,
    p_row_no         NUMBER,
    p_raw_line       CLOB,
    p_order_no_raw   VARCHAR2,
    p_order_date_raw VARCHAR2,
    p_customer_raw   VARCHAR2,
    p_amount_raw     VARCHAR2
  ) IS
  BEGIN
    INSERT INTO stg_order_import(
      batch_id, row_no, raw_line,
      order_no_raw, order_date_raw, customer_raw, amount_raw
    ) VALUES (
      p_batch_id, p_row_no, p_raw_line,
      p_order_no_raw, p_order_date_raw, p_customer_raw, p_amount_raw
    );
  END;
END;
/

CREATE OR REPLACE PACKAGE order_import_validate AS
  PROCEDURE validate_batch(p_batch_id NUMBER);
END;
/

CREATE OR REPLACE PACKAGE BODY order_import_validate AS
  FUNCTION is_number(p_value VARCHAR2) RETURN BOOLEAN IS
    v_number     NUMBER;
    v_normalized VARCHAR2(100);
  BEGIN
    v_normalized := REPLACE(TRIM(p_value), ',', '');

    IF NOT REGEXP_LIKE(v_normalized, '^-?[0-9]+([.][0-9]+)?$') THEN
      RETURN FALSE;
    END IF;

    v_number := TO_NUMBER(
      v_normalized,
      'FM999999999999999999999999999999D9999999999',
      'NLS_NUMERIC_CHARACTERS=.,'
    );
    RETURN TRUE;
  EXCEPTION
    WHEN VALUE_ERROR OR INVALID_NUMBER THEN
      RETURN FALSE;
  END;

  FUNCTION is_date_yyyymmdd(p_value VARCHAR2) RETURN BOOLEAN IS
    v_date DATE;
  BEGIN
    IF NOT REGEXP_LIKE(TRIM(p_value), '^[0-9]{8}$') THEN
      RETURN FALSE;
    END IF;

    v_date := TO_DATE(TRIM(p_value), 'FXYYYYMMDD');
    RETURN TRUE;
  EXCEPTION
    WHEN OTHERS THEN
      RETURN FALSE;
  END;

  PROCEDURE set_error(
    p_batch_id NUMBER,
    p_row_no   NUMBER,
    p_code     VARCHAR2,
    p_msg      VARCHAR2
  ) IS
  BEGIN
    UPDATE stg_order_import
       SET status = 'ERROR',
           error_code = p_code,
           error_msg = p_msg,
           updated_at = SYSTIMESTAMP
     WHERE batch_id = p_batch_id
       AND row_no = p_row_no;
  END;

  PROCEDURE validate_batch(p_batch_id NUMBER) IS
  BEGIN
    FOR r IN (
      SELECT *
        FROM stg_order_import
       WHERE batch_id = p_batch_id
         AND status IN ('RAW','ERROR')
       ORDER BY row_no
    ) LOOP
      IF r.order_no_raw IS NULL THEN
        set_error(p_batch_id, r.row_no, 'ORDER_NO_REQUIRED', 'order_no is required');
      ELSIF r.order_date_raw IS NULL OR NOT is_date_yyyymmdd(r.order_date_raw) THEN
        set_error(p_batch_id, r.row_no, 'ORDER_DATE_INVALID', 'order_date must be YYYYMMDD');
      ELSIF r.amount_raw IS NULL OR NOT is_number(r.amount_raw) THEN
        set_error(p_batch_id, r.row_no, 'AMOUNT_INVALID', 'amount must be number');
      ELSE
        UPDATE stg_order_import
           SET status = 'VALID',
               error_code = NULL,
               error_msg = NULL,
               updated_at = SYSTIMESTAMP
         WHERE batch_id = p_batch_id
           AND row_no = r.row_no;
      END IF;
    END LOOP;
  END;
END;
/

-- 顧客マスタに存在しない行をエラーにする
UPDATE stg_order_import s
   SET status = 'ERROR',
       error_code = 'CUSTOMER_NOT_FOUND',
       error_msg = 'customer code does not exist',
       updated_at = SYSTIMESTAMP
 WHERE s.batch_id = :batch_id
   AND s.status = 'VALID'
   AND NOT EXISTS (
     SELECT 1
       FROM customer_mst c
      WHERE c.customer_code = s.customer_raw
   );

-- 同じファイル内で注文番号が重複している行をエラーにする
UPDATE stg_order_import s
   SET status = 'ERROR',
       error_code = 'DUPLICATE_IN_FILE',
       error_msg = 'order_no is duplicated in source file',
       updated_at = SYSTIMESTAMP
 WHERE s.batch_id = :batch_id
   AND s.status = 'VALID'
   AND s.order_no_raw IN (
     SELECT order_no_raw
       FROM stg_order_import
      WHERE batch_id = :batch_id
        AND status = 'VALID'
      GROUP BY order_no_raw
     HAVING COUNT(*) > 1
   );

MERGE INTO orders t
USING (
  SELECT
    batch_id,
    row_no,
    order_no_raw AS order_no,
    TO_DATE(TRIM(order_date_raw), 'FXYYYYMMDD') AS order_date,
    customer_raw AS customer_code,
    TO_NUMBER(
      REPLACE(TRIM(amount_raw), ',', ''),
      'FM999999999999999999999999999999D9999999999',
      'NLS_NUMERIC_CHARACTERS=.,'
    ) AS amount
  FROM stg_order_import
  WHERE batch_id = :batch_id
    AND status = 'VALID'
) s
ON (t.order_no = s.order_no)
WHEN MATCHED THEN
  UPDATE SET
    t.order_date = s.order_date,
    t.customer_code = s.customer_code,
    t.amount = s.amount,
    t.updated_at = SYSTIMESTAMP
WHEN NOT MATCHED THEN
  INSERT (order_id, order_no, order_date, customer_code, amount, created_at, updated_at)
  VALUES (order_seq.NEXTVAL, s.order_no, s.order_date, s.customer_code, s.amount,
          SYSTIMESTAMP, SYSTIMESTAMP);

UPDATE stg_order_import s
   SET status = 'LOADED',
       loaded_order_id = (
         SELECT o.order_id
           FROM orders o
          WHERE o.order_no = s.order_no_raw
       ),
       updated_at = SYSTIMESTAMP
 WHERE s.batch_id = :batch_id
   AND s.status = 'VALID'
   AND EXISTS (
     SELECT 1
       FROM orders o
      WHERE o.order_no = s.order_no_raw
   );

UPDATE import_batch b
   SET total_count = (
         SELECT COUNT(*) FROM stg_order_import s WHERE s.batch_id = b.batch_id
       ),
       valid_count = (
         SELECT COUNT(*) FROM stg_order_import s WHERE s.batch_id = b.batch_id AND s.status IN ('VALID','LOADED')
       ),
       error_count = (
         SELECT COUNT(*) FROM stg_order_import s WHERE s.batch_id = b.batch_id AND s.status = 'ERROR'
       ),
       loaded_count = (
         SELECT COUNT(*) FROM stg_order_import s WHERE s.batch_id = b.batch_id AND s.status = 'LOADED'
       ),
       status = CASE
         WHEN EXISTS (
           SELECT 1 FROM stg_order_import s
            WHERE s.batch_id = b.batch_id
              AND s.status = 'ERROR'
         ) THEN 'ERROR'
         ELSE 'LOADED'
       END,
       finished_at = SYSTIMESTAMP
 WHERE b.batch_id = :batch_id;

-- 例: 金額のカンマを除去して再検証対象へ戻す
UPDATE stg_order_import
   SET amount_raw = REPLACE(amount_raw, ',', ''),
       status = 'RAW',
       error_code = NULL,
       error_msg = NULL,
       updated_at = SYSTIMESTAMP
 WHERE batch_id = :batch_id
   AND status = 'ERROR'
   AND error_code = 'AMOUNT_INVALID';

BEGIN
  order_import_validate.validate_batch(:batch_id);
END;
/

-- 再検証後、VALIDになった行だけ本登録へ進める

-- バッチ一覧
SELECT batch_id, source_name, status,
       total_count, valid_count, error_count, loaded_count,
       started_at, finished_at
  FROM import_batch
 ORDER BY batch_id DESC;

-- エラー行の確認
SELECT row_no, order_no_raw, order_date_raw, customer_raw, amount_raw,
       error_code, error_msg
  FROM stg_order_import
 WHERE batch_id = :batch_id
   AND status = 'ERROR'
 ORDER BY row_no;

-- まだ本登録されていない正常行
SELECT COUNT(*) AS valid_not_loaded
  FROM stg_order_import
 WHERE batch_id = :batch_id
   AND status = 'VALID';