概要

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

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

Snowflake の COPY INTO と Task で作るデータ取り込みパイプラインの実装例 #Snowflake - Qiita

事前準備

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

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;

テーブルを作成

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
);

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

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;
$$;

動作確認

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
);

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

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"
}

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

SELECT *
FROM ORDERS
ORDER BY ORDER_ID;

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"
}

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

SELECT *
FROM ORDERS
ORDER BY ORDER_ID;

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
);

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

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"
}

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

SELECT *
FROM ORDERS
WHERE ORDER_ID = 1;