0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@manabian

Snowflake のストアドプロシージャで Bronze → Silver の MERGE を実装する方法

0
Last updated at Posted at 2026-10-05

概要

Snowflake のタスクで Append した Bronze テーブルから、Silver テーブルへ MERGE でデータを連携する方法を紹介します。

Bronze テーブルの処理としては下記の記事で紹介しています。

事前準備

データベースとスキーマを作成

CREATE OR REPLACE DATABASE SILVER_TEST_DB;
CREATE OR REPLACE SCHEMA SILVER_TEST_DB.TEST_SCHEMA;

USE DATABASE SILVER_TEST_DB;
USE SCHEMA TEST_SCHEMA;

image.png

テーブルを作成

Bronze テーブルはすべての業務カラムを VARCHAR で定義し、Silver テーブルでは適切なデータ型で定義します。

CREATE OR REPLACE TABLE ORDERS_BRONZE (
    ORDER_ID                 VARCHAR,
    CUSTOMER_NAME            VARCHAR,
    AMOUNT                   VARCHAR,
    ORDER_DATE               VARCHAR,
    AUDIT__INGEST_TIMESTAMP  TIMESTAMP_LTZ,
    AUDIT__UPDATE_TIMESTAMP  TIMESTAMP_LTZ
);

CREATE OR REPLACE TABLE ORDERS (
    ORDER_ID                 NUMBER,
    CUSTOMER_NAME            VARCHAR,
    AMOUNT                   NUMBER(18, 2),
    ORDER_DATE               DATE,
    AUDIT__INGEST_TIMESTAMP  TIMESTAMP_LTZ,
    AUDIT__UPDATE_TIMESTAMP  TIMESTAMP_LTZ
);

image.png

ストアドプロシージャを定義

CREATE OR REPLACE PROCEDURE SP_UPSERT_BRONZE_TO_SILVER(
    P_TARGET_TABLE VARCHAR,
    P_SOURCE_SQL   VARCHAR,
    P_PRIMARY_KEYS ARRAY
)
RETURNS VARIANT
LANGUAGE SQL
EXECUTE AS OWNER
AS
$$
DECLARE
    max_update_timestamp   TIMESTAMP_LTZ;
    min_ingest_timestamp   TIMESTAMP_LTZ;

    min_ingest_sql         VARCHAR;
    on_condition           VARCHAR;
    partition_columns      VARCHAR;
    key_not_null_condition VARCHAR;
    merge_sql              VARCHAR;

    inserted_rows          NUMBER DEFAULT 0;
    updated_rows           NUMBER DEFAULT 0;
    affected_rows          NUMBER DEFAULT 0;

BEGIN
    -- =========================================================
    -- 1. MERGE 先の最新更新日時を取得
    -- =========================================================
    SELECT
        MAX(AUDIT__UPDATE_TIMESTAMP)
    INTO
        :max_update_timestamp
    FROM
        IDENTIFIER(:P_TARGET_TABLE);

    -- =========================================================
    -- 2. MERGE の再処理開始位置を取得
    --
    -- MERGE 先が空の場合:
    --   ソース SQL 全体の最小 INGEST 時刻
    --
    -- MERGE 先にデータがある場合:
    --   MERGE 先の MAX(AUDIT__UPDATE_TIMESTAMP) より後の
    --   AUDIT__UPDATE_TIMESTAMP を持つソースデータから、
    --   最小 AUDIT__INGEST_TIMESTAMP を取得
    -- =========================================================
    min_ingest_sql := '
        SELECT
            MIN(B.AUDIT__INGEST_TIMESTAMP)
                AS MIN_INGEST_TIMESTAMP
        FROM (
            ' || P_SOURCE_SQL || '
        ) AS B
    ';

    IF (max_update_timestamp IS NOT NULL) THEN
        min_ingest_sql := min_ingest_sql || '
            WHERE
                B.AUDIT__UPDATE_TIMESTAMP > ?
        ';
    END IF;

    LET min_ingest_result RESULTSET := (
        SELECT NULL::TIMESTAMP_LTZ AS MIN_INGEST_TIMESTAMP
    );

    IF (max_update_timestamp IS NULL) THEN
        min_ingest_result := (
            EXECUTE IMMEDIATE :min_ingest_sql
        );
    ELSE
        min_ingest_result := (
            EXECUTE IMMEDIATE :min_ingest_sql
            USING (
                max_update_timestamp
            )
        );
    END IF;

    FOR min_ingest_row IN min_ingest_result DO
        min_ingest_timestamp :=
            min_ingest_row.MIN_INGEST_TIMESTAMP;
    END FOR;

    -- =========================================================
    -- 3. 処理対象なし
    -- =========================================================
    IF (min_ingest_timestamp IS NULL) THEN
        RETURN OBJECT_CONSTRUCT(
            'status', 'NO_DATA_TO_MERGE',
            'max_update_timestamp', max_update_timestamp,
            'min_ingest_timestamp', min_ingest_timestamp,
            'affected_rows', 0
        );
    END IF;

    -- =========================================================
    -- 4. 主キー配列から SQL の各部分を生成
    --
    -- ・ON の結合条件
    -- ・PARTITION BY の列一覧
    -- ・主キーの NULL 除外条件
    -- =========================================================
    WITH KEY_COLUMNS AS (
        SELECT
            F.INDEX AS KEY_INDEX,
            '"' || REPLACE(F.VALUE::VARCHAR, '"', '""') || '"'
                AS COLUMN_NAME
        FROM
            TABLE(FLATTEN(INPUT => :P_PRIMARY_KEYS)) AS F
    )
    SELECT
        LISTAGG(
            'T.' || COLUMN_NAME || ' = S.' || COLUMN_NAME,
            ' AND '
        ) WITHIN GROUP (ORDER BY KEY_INDEX),

        LISTAGG(
            'B.' || COLUMN_NAME,
            ', '
        ) WITHIN GROUP (ORDER BY KEY_INDEX),

        LISTAGG(
            'B.' || COLUMN_NAME || ' IS NOT NULL',
            ' AND '
        ) WITHIN GROUP (ORDER BY KEY_INDEX)
    INTO
        :on_condition,
        :partition_columns,
        :key_not_null_condition
    FROM
        KEY_COLUMNS;

    -- =========================================================
    -- 5. MERGE SQL を生成
    --
    -- ソース SQL から再処理範囲を抽出
    -- 主キーに NULL を含む行を除外
    -- 同じ主キーが複数ある場合は最新の１行を採用
    -- =========================================================
    merge_sql := '
        MERGE INTO IDENTIFIER(?) AS T

        USING (
            SELECT
                B.*
            FROM (
                ' || P_SOURCE_SQL || '
            ) AS B
            WHERE
                B.AUDIT__INGEST_TIMESTAMP >= ?
                AND ' || key_not_null_condition || '
            QUALIFY
                ROW_NUMBER() OVER (
                    PARTITION BY
                        ' || partition_columns || '
                    ORDER BY
                        B.AUDIT__INGEST_TIMESTAMP DESC NULLS LAST,
                        B.AUDIT__UPDATE_TIMESTAMP DESC NULLS LAST
                ) = 1
        ) AS S

        ON
            ' || on_condition || '

        WHEN MATCHED
            AND (
                T.AUDIT__UPDATE_TIMESTAMP IS NULL
                OR S.AUDIT__UPDATE_TIMESTAMP
                    > T.AUDIT__UPDATE_TIMESTAMP
            )
            THEN UPDATE ALL BY NAME

        WHEN NOT MATCHED THEN
            INSERT ALL BY NAME
    ';

    -- =========================================================
    -- 6. MERGE を実行
    --
    -- テーブル名と開始日時をバインド
    -- RESULTSET は実行結果で初期化
    -- =========================================================
    LET merge_result RESULTSET := (
        EXECUTE IMMEDIATE :merge_sql
        USING (
            P_TARGET_TABLE,
            min_ingest_timestamp
        )
    );

    -- =========================================================
    -- 7. 実行結果から処理件数を取得
    -- =========================================================
    FOR merge_row IN merge_result DO
        inserted_rows := merge_row."number of rows inserted";
        updated_rows  := merge_row."number of rows updated";
    END FOR;

    affected_rows := inserted_rows + updated_rows;

    RETURN OBJECT_CONSTRUCT(
        'status', 'SUCCESS',
        'max_update_timestamp', max_update_timestamp,
        'min_ingest_timestamp', min_ingest_timestamp,
        'affected_rows', affected_rows
    );
END;
$$;

image.png

動作確認

1. INSERT の実施

Bronze テーブルにデータを挿入します。

INSERT INTO ORDERS_BRONZE (
    ORDER_ID,
    CUSTOMER_NAME,
    AMOUNT,
    ORDER_DATE,
    AUDIT__INGEST_TIMESTAMP,
    AUDIT__UPDATE_TIMESTAMP
)
VALUES
(
    '1',
    'Alice',
    '100.00',
    '2026-10-01',
    '2026-10-04 10:00:00'::TIMESTAMP_LTZ,
    '2026-10-04 10:00:00'::TIMESTAMP_LTZ
),
(
    '2',
    'Bob',
    '200.00',
    '2026-10-02',
    '2026-10-04 10:00:00'::TIMESTAMP_LTZ,
    '2026-10-04 10:00:00'::TIMESTAMP_LTZ
);

image.png

ストアドプロシージャを実行します。

CALL SP_UPSERT_BRONZE_TO_SILVER(
    P_TARGET_TABLE => 'COPY_TEST_DB.TEST_SCHEMA.ORDERS',

    P_SOURCE_SQL => $$
        SELECT
            TRY_CAST(ORDER_ID AS NUMBER) AS ORDER_ID,
            CUSTOMER_NAME,
            TRY_CAST(AMOUNT AS NUMBER(18, 2)) AS AMOUNT,
            TRY_CAST(ORDER_DATE AS DATE) AS ORDER_DATE,
            AUDIT__INGEST_TIMESTAMP,
            AUDIT__UPDATE_TIMESTAMP
        FROM
            COPY_TEST_DB.TEST_SCHEMA.ORDERS_BRONZE
    $$,

    P_PRIMARY_KEYS => ARRAY_CONSTRUCT('ORDER_ID')
);

{
  "affected_rows": 2,
  "min_ingest_timestamp": "2026-10-04 10:00:00.000 -0700",
  "status": "SUCCESS"
}

image.png

データが挿入されていることを確認します。

SELECT *
FROM ORDERS
ORDER BY ORDER_ID;

image.png

2. UPSERT の実施

ORDER_ID が 1 のデータを更新対象、3 のデータを挿入対象として、Bronze テーブルにデータを挿入します。

INSERT INTO ORDERS_BRONZE (
    ORDER_ID,
    CUSTOMER_NAME,
    AMOUNT,
    ORDER_DATE,
    AUDIT__INGEST_TIMESTAMP,
    AUDIT__UPDATE_TIMESTAMP
)
VALUES
(
    '1',
    'Alice Updated',
    '150.00',
    '2026-10-01',
    '2026-10-04 11:00:00'::TIMESTAMP_LTZ,
    '2026-10-04 11:00:00'::TIMESTAMP_LTZ
),
(
    '3',
    'Charlie',
    '300.00',
    '2026-10-03',
    '2026-10-04 11:00:00'::TIMESTAMP_LTZ,
    '2026-10-04 11:00:00'::TIMESTAMP_LTZ
);

ストアドプロシージャを実行します。

CALL SP_UPSERT_BRONZE_TO_SILVER(
    P_TARGET_TABLE => 'COPY_TEST_DB.TEST_SCHEMA.ORDERS',

    P_SOURCE_SQL => $$
        SELECT
            TRY_CAST(ORDER_ID AS NUMBER) AS ORDER_ID,
            CUSTOMER_NAME,
            TRY_CAST(AMOUNT AS NUMBER(18, 2)) AS AMOUNT,
            TRY_CAST(ORDER_DATE AS DATE) AS ORDER_DATE,
            AUDIT__INGEST_TIMESTAMP,
            AUDIT__UPDATE_TIMESTAMP
        FROM
            COPY_TEST_DB.TEST_SCHEMA.ORDERS_BRONZE
    $$,

    P_PRIMARY_KEYS => ARRAY_CONSTRUCT('ORDER_ID')
);

{
  "affected_rows": 2,
  "max_update_timestamp": "2026-10-04 10:00:00.000 -0700",
  "min_ingest_timestamp": "2026-10-04 11:00:00.000 -0700",
  "status": "SUCCESS"
}

image.png

データが UPSERT されていることを確認します。

SELECT *
FROM ORDERS
ORDER BY ORDER_ID;

image.png

3. 遅延データの処理

AUDIT__INGEST_TIMESTAMP が 2026-10-04 11:00:00.000 -0700 より前の遅延データを挿入します。AUDIT__UPDATE_TIMESTAMP は既存データより新しい値ですが、同一 ORDER_ID 内では AUDIT__INGEST_TIMESTAMP が最新のデータを優先するため、このデータは反映されない想定です。

INSERT INTO ORDERS_BRONZE (
    ORDER_ID,
    CUSTOMER_NAME,
    AMOUNT,
    ORDER_DATE,
    AUDIT__INGEST_TIMESTAMP,
    AUDIT__UPDATE_TIMESTAMP
)
VALUES
(
    '1',
    'Alice Old Data',
    '120.00',
    '2026-10-01',
    '2026-10-04 10:30:00'::TIMESTAMP_LTZ,
    '2026-10-04 11:30:00'::TIMESTAMP_LTZ
);

image.png

ストアドプロシージャを実行します。

CALL SP_UPSERT_BRONZE_TO_SILVER(
    P_TARGET_TABLE => 'COPY_TEST_DB.TEST_SCHEMA.ORDERS',

    P_SOURCE_SQL => $$
        SELECT
            TRY_CAST(ORDER_ID AS NUMBER) AS ORDER_ID,
            CUSTOMER_NAME,
            TRY_CAST(AMOUNT AS NUMBER(18, 2)) AS AMOUNT,
            TRY_CAST(ORDER_DATE AS DATE) AS ORDER_DATE,
            AUDIT__INGEST_TIMESTAMP,
            AUDIT__UPDATE_TIMESTAMP
        FROM
            COPY_TEST_DB.TEST_SCHEMA.ORDERS_BRONZE
    $$,

    P_PRIMARY_KEYS => ARRAY_CONSTRUCT('ORDER_ID')
);

{
  "affected_rows": 0,
  "max_update_timestamp": "2026-10-04 11:00:00.000 -0700",
  "min_ingest_timestamp": "2026-10-04 10:30:00.000 -0700",
  "status": "SUCCESS"
}

image.png

データが更新されていないことを確認します。

SELECT *
FROM ORDERS
WHERE ORDER_ID = 1;

image.png

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?