【PL/SQL】パイプライン関数で大量データ処理を勝たせる完全ガイド｜CARDINALITY・並列ETL・BULK内蔵・チェーン設計

-- ① 行の型
CREATE OR REPLACE TYPE t_sales_row AS OBJECT(
  sales_id   NUMBER,
  product_cd VARCHAR2(20),
  amount     NUMBER
);
/

-- ② コレクション型
CREATE OR REPLACE TYPE t_sales_tab IS TABLE OF t_sales_row;
/

-- ③ パイプライン関数
CREATE OR REPLACE FUNCTION f_sales_stream
  RETURN t_sales_tab PIPELINED
AS
BEGIN
  FOR rec IN (SELECT sales_id, product_cd, amount FROM raw_sales) LOOP
    PIPE ROW(t_sales_row(rec.sales_id, rec.product_cd, rec.amount));
  END LOOP;
  RETURN;  -- 値なしで終了（必須）
END;
/

-- FROM句で呼び出す
SELECT *
  FROM TABLE(f_sales_stream())
 WHERE amount > 10000;

-- ❌ 何もしないとオプティマイザは8192行と決め打ち
SELECT /*+ LEADING(t f) USE_NL(f) */ *
  FROM base_table t
  JOIN TABLE(f_sales_stream()) f
    ON (t.product_cd = f.product_cd);
-- → NESTED LOOPSで数時間コース

-- ✅ 方法1: CARDINALITYヒント（簡単）
SELECT /*+ CARDINALITY(f 10000000) */ *
  FROM base_table t,
       TABLE(f_sales_stream()) f
 WHERE t.product_cd = f.product_cd;

-- ✅ 方法2: DYNAMIC_SAMPLINGで動的統計収集（2〜4の値推奨）
SELECT /*+ DYNAMIC_SAMPLING(f 4) */ *
  FROM base_table t,
       TABLE(f_sales_stream()) f
 WHERE t.product_cd = f.product_cd;

-- ✅ 方法3: 関数内に OPTIMIZER_FOR_HINTING（12c以降）
-- または ASSOCIATE STATISTICS WITH FUNCTIONS でコストモデルを登録
ASSOCIATE STATISTICS WITH FUNCTIONS f_sales_stream
  DEFAULT SELECTIVITY 10,
  DEFAULT COST (1000000, 10, 0);

CREATE OR REPLACE FUNCTION f_mask_sales(
  p_cur IN SYS_REFCURSOR
)
RETURN t_sales_tab
PIPELINED
PARALLEL_ENABLE(PARTITION p_cur BY ANY)   -- ← 並列化を許可
AS
  rec raw_sales%ROWTYPE;
BEGIN
  LOOP
    FETCH p_cur INTO rec;
    EXIT WHEN p_cur%NOTFOUND;

    -- 行単位の変換処理（マスキング等）
    PIPE ROW(t_sales_row(
      rec.sales_id,
      CASE WHEN rec.customer_id IS NOT NULL
           THEN 'MASK-' || MOD(rec.customer_id, 10000)
           ELSE rec.product_cd END,
      rec.amount
    ));
  END LOOP;
  CLOSE p_cur;
  RETURN;
END;
/

-- 呼び出し時に並列度4を指定
SELECT /*+ PARALLEL(f, 4) CARDINALITY(f 10000000) */ *
  FROM TABLE(f_mask_sales(
    CURSOR(SELECT /*+ PARALLEL(s, 4) */ * FROM raw_sales s)
  )) f;

CREATE OR REPLACE FUNCTION f_sales_fast(
  p_cur IN SYS_REFCURSOR
)
RETURN t_sales_tab PIPELINED
AS
  TYPE t_rec IS TABLE OF raw_sales%ROWTYPE;
  v_batch t_rec;
  C_LIMIT CONSTANT PLS_INTEGER := 1000;  -- 100〜5000の範囲が実用域
BEGIN
  LOOP
    FETCH p_cur BULK COLLECT INTO v_batch LIMIT C_LIMIT;
    EXIT WHEN v_batch.COUNT = 0;

    FOR i IN 1 .. v_batch.COUNT LOOP
      PIPE ROW(t_sales_row(
        v_batch(i).sales_id,
        v_batch(i).product_cd,
        v_batch(i).amount * 1.1  -- 例: 10%加算
      ));
    END LOOP;
  END LOOP;
  CLOSE p_cur;
  RETURN;
END;
/

-- ステージ1: 抽出（日付フィルタ）
CREATE OR REPLACE FUNCTION f_extract(p_cur SYS_REFCURSOR)
  RETURN t_sales_tab PIPELINED PARALLEL_ENABLE(PARTITION p_cur BY ANY) AS
  rec raw_sales%ROWTYPE;
BEGIN
  LOOP FETCH p_cur INTO rec; EXIT WHEN p_cur%NOTFOUND;
    IF rec.sales_date BETWEEN DATE '2026-01-01' AND DATE '2026-12-31' THEN
      PIPE ROW(t_sales_row(rec.sales_id, rec.product_cd, rec.amount));
    END IF;
  END LOOP;
  CLOSE p_cur; RETURN;
END;
/

-- ステージ2: 変換（税込計算）
CREATE OR REPLACE FUNCTION f_transform(p_cur SYS_REFCURSOR)
  RETURN t_sales_tab PIPELINED PARALLEL_ENABLE(PARTITION p_cur BY ANY) AS
  rec t_sales_row;
BEGIN
  LOOP FETCH p_cur INTO rec.sales_id, rec.product_cd, rec.amount;
       EXIT WHEN p_cur%NOTFOUND;
    rec.amount := rec.amount * 1.10;
    PIPE ROW(rec);
  END LOOP;
  CLOSE p_cur; RETURN;
END;
/

-- ステージ3: マスキング
CREATE OR REPLACE FUNCTION f_mask(p_cur SYS_REFCURSOR)
  RETURN t_sales_tab PIPELINED PARALLEL_ENABLE(PARTITION p_cur BY ANY) AS
  rec t_sales_row;
BEGIN
  LOOP FETCH p_cur INTO rec.sales_id, rec.product_cd, rec.amount;
       EXIT WHEN p_cur%NOTFOUND;
    rec.product_cd := 'PRD-' || MOD(rec.sales_id, 1000);
    PIPE ROW(rec);
  END LOOP;
  CLOSE p_cur; RETURN;
END;
/

-- 3段チェーンを1SQL で実行
INSERT /*+ APPEND */ INTO staged_sales
SELECT *
  FROM TABLE(f_mask(
         CURSOR(SELECT * FROM TABLE(f_transform(
           CURSOR(SELECT * FROM TABLE(f_extract(
             CURSOR(SELECT * FROM raw_sales)
           )))
         )))
       ));

-- 手法1: DBMS_OUTPUTで進捗ログ（開発時）
CREATE OR REPLACE FUNCTION f_with_log(p_cur SYS_REFCURSOR)
  RETURN t_sales_tab PIPELINED AS
  rec raw_sales%ROWTYPE;
  v_cnt NUMBER := 0;
BEGIN
  LOOP FETCH p_cur INTO rec; EXIT WHEN p_cur%NOTFOUND;
    v_cnt := v_cnt + 1;
    IF MOD(v_cnt, 100000) = 0 THEN
      DBMS_OUTPUT.PUT_LINE('Processed: ' || v_cnt || ' / ts=' || SYSTIMESTAMP);
    END IF;
    PIPE ROW(t_sales_row(rec.sales_id, rec.product_cd, rec.amount));
  END LOOP;
  CLOSE p_cur; RETURN;
END;
/

-- 手法2: 自己診断表にAUTONOMOUS_TRANSACTIONで書き込む（本番可）
CREATE TABLE pipeline_diag(
  ts     TIMESTAMP,
  fn_nm  VARCHAR2(64),
  phase  VARCHAR2(32),
  cnt    NUMBER,
  info   VARCHAR2(4000)
);

CREATE OR REPLACE PROCEDURE log_diag(
  p_fn VARCHAR2, p_phase VARCHAR2, p_cnt NUMBER, p_info VARCHAR2
) AS
  PRAGMA AUTONOMOUS_TRANSACTION;
BEGIN
  INSERT INTO pipeline_diag VALUES(SYSTIMESTAMP, p_fn, p_phase, p_cnt, p_info);
  COMMIT;
END;
/

-- 手法3: SQLトレース（10046イベント）
ALTER SESSION SET EVENTS '10046 trace name context forever, level 12';
-- ここでパイプライン関数を呼ぶSQLを実行
ALTER SESSION SET EVENTS '10046 trace name context off';
-- トレースファイルを tkprof で整形して分析