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

Snowflake

Last updated at 2026-10-09Posted at 2026-10-04

概要

Snowflake にて、COPY INTO をベースにしたデータ取り込み処理の実装方法を紹介します。

下記の処理を実施します。

COPY INTO によりファイルからテーブルにデータをロードする処理
COPY INTO の処理結果を確認する処理（パラメータに応じて、エラーとするかメッセージ出力のみとするかを切り替え）
指定期間より前に取り込まれたデータをテーブルから削除する処理

テーブルのカラムのデータ型を文字型としており、メダリオンアーキテクチャの Bronze テーブルに対する処理を想定しています。

事前準備

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

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

USE DATABASE COPY_TEST_DB;
USE SCHEMA TEST_SCHEMA;

テーブルを作成

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

ステージを作成

CREATE OR REPLACE STAGE INPUT_STAGE;

ファイルを配置する関数を定義

from io import BytesIO

from snowflake.snowpark.context import get_active_session

session = get_active_session()

from pathlib import Path

def reset_test_files():
    session.sql(
        "REMOVE @INPUT_STAGE/test_files"
    ).collect()


def put_csv(file_name, content):
    local_path = f"/tmp/{file_name}"

    Path(local_path).write_text(
        content,
        encoding="utf-8"
    )

    session.file.put(
        local_path,
        "@INPUT_STAGE/test_files",
        auto_compress=False,
        overwrite=True
    )

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

COPY INTO によりファイルからテーブルにデータをロードするストアドプロシージャを定義

CREATE OR REPLACE PROCEDURE SP_COPY(
    P_TARGET_TABLE         VARCHAR,
    P_SOURCE_STAGE         VARCHAR,
    P_SOURCE_PATH          VARCHAR DEFAULT NULL,
    P_FILE_FORMAT          VARCHAR DEFAULT NULL,
    P_MATCH_BY_COLUMN_NAME VARCHAR DEFAULT NULL,
    P_INCLUDE_METADATA     VARCHAR DEFAULT NULL,
    P_ON_ERROR             VARCHAR DEFAULT NULL,
    P_FILES                VARCHAR DEFAULT NULL,
    P_PATTERN              VARCHAR DEFAULT NULL,
    P_FORCE                BOOLEAN DEFAULT NULL,
    P_PURGE                BOOLEAN DEFAULT NULL,
    P_SIZE_LIMIT           NUMBER  DEFAULT NULL,
    P_LOAD_UNCERTAIN_FILES BOOLEAN DEFAULT NULL,
    P_ENFORCE_LENGTH       BOOLEAN DEFAULT NULL
)
RETURNS VARCHAR
LANGUAGE SQL
EXECUTE AS OWNER
AS
$$
DECLARE
    copy_sql                    VARCHAR;
    stage_location              VARCHAR;

    copy_query_id               VARCHAR;
    copy_status                 VARCHAR;
    load_status                 VARCHAR;

    processed_file_count        NUMBER DEFAULT 0;
    loaded_file_count           NUMBER DEFAULT 0;
    load_failed_file_count      NUMBER DEFAULT 0;
    partially_loaded_file_count NUMBER DEFAULT 0;

    rows_parsed                 NUMBER DEFAULT 0;
    rows_loaded                 NUMBER DEFAULT 0;
    errors_seen                 NUMBER DEFAULT 0;

    return_value                VARCHAR;

BEGIN
    -- =========================================================
    -- ステージパスを生成
    -- P_SOURCE_STAGE は @ を含めずに指定
    -- =========================================================
    stage_location := '@' || P_SOURCE_STAGE;

    IF (P_SOURCE_PATH IS NOT NULL AND P_SOURCE_PATH <> '') THEN
        stage_location := stage_location || '/' ||
            LTRIM(P_SOURCE_PATH, '/');
    END IF;

    -- =========================================================
    -- COPY INTO の基本SQLを生成
    -- ステージパスは空白などを扱えるよう一重引用符で囲む
    -- =========================================================
    copy_sql :=
        'COPY INTO IDENTIFIER(?) ' ||
        'FROM ''' ||
            REPLACE(stage_location, '''', '''''') ||
        ''' ';

    -- =========================================================
    -- 対象ファイル
    -- P_FILES は括弧を含まないSQL断片
    -- 例: 'orders_01.csv', 'orders_02.csv'
    -- =========================================================
    IF (P_FILES IS NOT NULL AND P_FILES <> '') THEN
        copy_sql := copy_sql ||
            'FILES = (' || P_FILES || ') ';
    END IF;

    -- =========================================================
    -- ファイル名の正規表現
    -- P_PATTERN は外側の引用符を含めずに指定
    -- =========================================================
    IF (P_PATTERN IS NOT NULL AND P_PATTERN <> '') THEN
        copy_sql := copy_sql ||
            'PATTERN = ''' ||
                REPLACE(P_PATTERN, '''', '''''') ||
            ''' ';
    END IF;

    -- =========================================================
    -- ファイル形式
    -- P_FILE_FORMAT は括弧を含まないSQL断片
    -- =========================================================
    IF (P_FILE_FORMAT IS NOT NULL AND P_FILE_FORMAT <> '') THEN
        copy_sql := copy_sql ||
            'FILE_FORMAT = (' || P_FILE_FORMAT || ') ';
    END IF;

    -- =========================================================
    -- 列名による対応付け
    -- =========================================================
    IF (
        P_MATCH_BY_COLUMN_NAME IS NOT NULL
        AND P_MATCH_BY_COLUMN_NAME <> ''
    ) THEN
        copy_sql := copy_sql ||
            'MATCH_BY_COLUMN_NAME = ''' ||
                REPLACE(P_MATCH_BY_COLUMN_NAME, '''', '''''') ||
            ''' ';
    END IF;

    -- =========================================================
    -- メタデータのマッピング
    -- P_INCLUDE_METADATA は括弧を含まないSQL断片
    -- =========================================================
    IF (
        P_INCLUDE_METADATA IS NOT NULL
        AND P_INCLUDE_METADATA <> ''
    ) THEN
        copy_sql := copy_sql ||
            'INCLUDE_METADATA = (' || P_INCLUDE_METADATA || ') ';
    END IF;

    -- =========================================================
    -- エラー時の動作
    -- =========================================================
    IF (P_ON_ERROR IS NOT NULL AND P_ON_ERROR <> '') THEN
        copy_sql := copy_sql ||
            'ON_ERROR = ''' ||
                REPLACE(P_ON_ERROR, '''', '''''') ||
            ''' ';
    END IF;

    -- =========================================================
    -- ロード済みファイルの強制再ロード
    -- =========================================================
    IF (P_FORCE IS NOT NULL) THEN
        copy_sql := copy_sql ||
            'FORCE = ' || IFF(P_FORCE, 'TRUE', 'FALSE') || ' ';
    END IF;

    -- =========================================================
    -- 正常ロード後のファイル削除
    -- =========================================================
    IF (P_PURGE IS NOT NULL) THEN
        copy_sql := copy_sql ||
            'PURGE = ' || IFF(P_PURGE, 'TRUE', 'FALSE') || ' ';
    END IF;

    -- =========================================================
    -- 1回のロード量の制限（バイト）
    -- =========================================================
    IF (P_SIZE_LIMIT IS NOT NULL) THEN
        copy_sql := copy_sql ||
            'SIZE_LIMIT = ' || TO_VARCHAR(P_SIZE_LIMIT) || ' ';
    END IF;

    -- =========================================================
    -- ロード状態が不明なファイルの取り込み
    -- =========================================================
    IF (P_LOAD_UNCERTAIN_FILES IS NOT NULL) THEN
        copy_sql := copy_sql ||
            'LOAD_UNCERTAIN_FILES = ' ||
                IFF(P_LOAD_UNCERTAIN_FILES, 'TRUE', 'FALSE') ||
            ' ';
    END IF;

    -- =========================================================
    -- 列の長さを超える文字列の扱い
    -- TRUE: エラー / FALSE: 切り捨て
    -- =========================================================
    IF (P_ENFORCE_LENGTH IS NOT NULL) THEN
        copy_sql := copy_sql ||
            'ENFORCE_LENGTH = ' ||
                IFF(P_ENFORCE_LENGTH, 'TRUE', 'FALSE') ||
            ' ';
    END IF;

    -- =========================================================
    -- 集計のため、成功ファイルも実行結果に含める
    -- =========================================================
    copy_sql := copy_sql ||
        'RETURN_FAILED_ONLY = FALSE';

    -- =========================================================
    -- データをロードし、実行直後に Query ID を保存
    -- =========================================================
    EXECUTE IMMEDIATE :copy_sql USING (P_TARGET_TABLE);
    copy_query_id := SQLID;

    -- =========================================================
    -- 0 files processed の場合は列構成が異なるため、
    -- 最初に status のみ確認
    -- =========================================================
    SELECT "status"
    INTO :copy_status
    FROM TABLE(RESULT_SCAN(:copy_query_id))
    LIMIT 1;

    -- =========================================================
    -- COPY INTO の実行結果を集計
    -- =========================================================
    IF (copy_status != 'Copy executed with 0 files processed.') THEN
        SELECT
            COUNT(*),
            COALESCE(COUNT_IF("status" = 'LOADED'), 0),
            COALESCE(COUNT_IF("status" = 'LOAD_FAILED'), 0),
            COALESCE(COUNT_IF("status" = 'PARTIALLY_LOADED'), 0),
            COALESCE(SUM("rows_parsed"), 0),
            COALESCE(SUM("rows_loaded"), 0),
            COALESCE(SUM("errors_seen"), 0)
        INTO
            :processed_file_count,
            :loaded_file_count,
            :load_failed_file_count,
            :partially_loaded_file_count,
            :rows_parsed,
            :rows_loaded,
            :errors_seen
        FROM TABLE(RESULT_SCAN(:copy_query_id));
    END IF;

    -- =========================================================
    -- ロード結果を判定
    -- =========================================================
    IF (load_failed_file_count > 0) THEN
        load_status := 'ERROR';
    ELSEIF (partially_loaded_file_count > 0) THEN
        load_status := 'PARTIAL_ERROR';
    ELSE
        load_status := 'SUCCESS';
    END IF;

    -- =========================================================
    -- 実行結果を JSON として返却
    -- =========================================================
    return_value := TO_JSON(
        OBJECT_CONSTRUCT(
            'load_status', load_status,
            'copy_query_id', copy_query_id,
            'processed_file_count', processed_file_count,
            'loaded_file_count', loaded_file_count,
            'load_failed_file_count', load_failed_file_count,
            'partially_loaded_file_count', partially_loaded_file_count,
            'rows_parsed', rows_parsed,
            'rows_loaded', rows_loaded,
            'errors_seen', errors_seen
        )
    );

    RETURN return_value;

EXCEPTION
    WHEN OTHER THEN
        RETURN TO_JSON(
            OBJECT_CONSTRUCT_KEEP_NULL(
                'load_status', 'ERROR',
                'copy_query_id', copy_query_id,
                'processed_file_count', processed_file_count,
                'loaded_file_count', loaded_file_count,
                'load_failed_file_count', load_failed_file_count,
                'partially_loaded_file_count', partially_loaded_file_count,
                'rows_parsed', rows_parsed,
                'rows_loaded', rows_loaded,
                'errors_seen', errors_seen,
                'error_code', SQLCODE,
                'error_message', SQLERRM,
                'sql_state', SQLSTATE
            )
        );
END;
$$;

COPY INTO の処理結果を確認するストアドプロシージャを定義

CREATE OR REPLACE PROCEDURE SP_CHECK_COPY_RESULT(
    copy_result VARCHAR,
    raise_load_failed_error BOOLEAN DEFAULT TRUE,
    raise_partially_loaded_error BOOLEAN DEFAULT FALSE
)
RETURNS VARCHAR
LANGUAGE SQL
EXECUTE AS OWNER
AS
$$
DECLARE
    result                       VARIANT;
    copy_query_id                VARCHAR;
    load_failed_file_count       NUMBER;
    partially_loaded_file_count  NUMBER;

    copy_load_error EXCEPTION (
        -20001,
        'COPY INTO detected LOAD_FAILED files.'
    );

    copy_partial_error EXCEPTION (
        -20002,
        'COPY INTO detected PARTIALLY_LOADED files.'
    );

BEGIN

    result := PARSE_JSON(copy_result);

    copy_query_id :=
        result:"copy_query_id"::VARCHAR;

    load_failed_file_count :=
        result:"load_failed_file_count"::NUMBER;

    partially_loaded_file_count :=
        result:"partially_loaded_file_count"::NUMBER;

    -- =========================================================
    -- LOAD_FAILED を確認
    -- =========================================================
    IF (load_failed_file_count > 0) THEN

        IF (raise_load_failed_error) THEN
            RAISE copy_load_error;
        END IF;

        RETURN
            'COPY INTO completed with LOAD_FAILED files. '
            || 'Load failed file count: '
            || load_failed_file_count
            || '. Query ID: '
            || copy_query_id;

    END IF;

    -- =========================================================
    -- PARTIALLY_LOADED を確認
    -- =========================================================
    IF (partially_loaded_file_count > 0) THEN

        IF (raise_partially_loaded_error) THEN
            RAISE copy_partial_error;
        END IF;

        RETURN
            'COPY INTO completed with PARTIALLY_LOADED files. '
            || 'Partially loaded file count: '
            || partially_loaded_file_count
            || '. Query ID: '
            || copy_query_id;

    END IF;

    -- =========================================================
    -- 正常終了
    -- =========================================================
    RETURN
        'COPY INTO completed successfully. '
        || 'Query ID: '
        || copy_query_id;

END;
$$;

処理結果を確認する処理を別のストアドプロシージャに分けたのは、COPY INTO の Query ID を後続処理で利用したかったためです。

ただし、現時点では、出力された Query ID を指定して実行結果を確認できていません。

SELECT
    *
FROM
    TABLE(
        RESULT_SCAN(
            '01c77fdd-0001-6b93-0001-48f20004c34e'  -- ToDo ここを SP_COPY_ORDERS の出力結果に書き換える
        )
    );

Statement 01c77fdd-0001-6b93-0001-48f20004c34e not found

指定期間より前に取り込まれたデータをテーブルから削除するストアドプロシージャを定義

CREATE OR REPLACE PROCEDURE SP_DELETE_OLD_ORDERS(
    retention_days NUMBER DEFAULT 60
)
RETURNS VARCHAR
LANGUAGE SQL
EXECUTE AS OWNER
AS
$$
DECLARE
    deleted_row_count NUMBER DEFAULT 0;

BEGIN

    DELETE FROM ORDERS
    WHERE
        AUDIT__INGEST_TIMESTAMP
            < DATEADD(
                DAY,
                -1 * :retention_days,
                CURRENT_TIMESTAMP()
            );

    deleted_row_count := SQLROWCOUNT;

    RETURN
        'Deleted '
        || deleted_row_count
        || ' rows older than '
        || retention_days
        || ' days.';

END;
$$;

動作確認

正常ファイル取り込み時の動作確認

ファイルを配置します。

normal_csv = """order_id,customer_name,amount,order_date
1,Alice,100.50,2026-10-04
2,Bob,200.00,2026-10-04
"""

reset_test_files()

put_csv(
    "normal_001.csv",
    normal_csv
)

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

-- METADATA$FILE_LAST_MODIFIED と METADATA$START_SCAN_TIME のタイムゾーンを統一するために必要な設定
ALTER SESSION SET TIMEZONE = 'UTC';

CALL SP_COPY(
    P_TARGET_TABLE => 'COPY_TEST_DB.TEST_SCHEMA.ORDERS',
    P_SOURCE_STAGE => 'COPY_TEST_DB.TEST_SCHEMA.INPUT_STAGE',
    P_SOURCE_PATH  => 'test_files/',
    P_FILE_FORMAT => $$
        TYPE = CSV
        PARSE_HEADER = TRUE
        FIELD_OPTIONALLY_ENCLOSED_BY = '"'
        ERROR_ON_COLUMN_COUNT_MISMATCH = FALSE
    $$,
    P_MATCH_BY_COLUMN_NAME => 'CASE_INSENSITIVE',
    P_INCLUDE_METADATA => $$
        AUDIT__INGEST_TIMESTAMP = METADATA$FILE_LAST_MODIFIED,
        AUDIT__UPDATE_TIMESTAMP = METADATA$START_SCAN_TIME
    $$,
    P_ON_ERROR => 'SKIP_FILE'
);

{"copy_query_id":"01c79bb4-0001-7e01-0001-48f20009d222","errors_seen":0,"load_failed_file_count":0,"load_status":"SUCCESS","loaded_file_count":1,"partially_loaded_file_count":0,"processed_file_count":1,"rows_loaded":2,"rows_parsed":2}

処理結果を確認します。

CALL SP_CHECK_COPY_RESULT(
    '{"copy_query_id":"01c79bb4-0001-7e01-0001-48f20009d222","errors_seen":0,"load_failed_file_count":0,"load_status":"SUCCESS","loaded_file_count":1,"partially_loaded_file_count":0,"processed_file_count":1,"rows_loaded":2,"rows_parsed":2}',  -- ToDo SP_COPY_ORDERS のリターン結果に書き換える
    TRUE
);

COPY INTO completed successfully. Query ID: 01c79bb4-0001-7e01-0001-48f20009d222

異常ファイル取り込み時の動作確認

ファイルを配置します。

invalid_csv = """order_id,customer_name,amount,order_date
3,Carol,300.00,2026-10-04
4,David,400.00,2026-10-04,EXTRA_COLUMN
"""

reset_test_files()

put_csv(
    "invalid_001.csv",
    invalid_csv
)

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

-- METADATA$FILE_LAST_MODIFIED と METADATA$START_SCAN_TIME のタイムゾーンを統一するために必要な設定
ALTER SESSION SET TIMEZONE = 'UTC';

CALL SP_COPY(
    P_TARGET_TABLE => 'COPY_TEST_DB.TEST_SCHEMA.ORDERS',
    P_SOURCE_STAGE => 'COPY_TEST_DB.TEST_SCHEMA.INPUT_STAGE',
    P_SOURCE_PATH  => 'test_files/',
    P_FILE_FORMAT => $$
        TYPE = CSV
        PARSE_HEADER = TRUE
        FIELD_OPTIONALLY_ENCLOSED_BY = '"'
        ERROR_ON_COLUMN_COUNT_MISMATCH = FALSE
    $$,
    P_MATCH_BY_COLUMN_NAME => 'CASE_INSENSITIVE',
    P_INCLUDE_METADATA => $$
        AUDIT__INGEST_TIMESTAMP = METADATA$FILE_LAST_MODIFIED,
        AUDIT__UPDATE_TIMESTAMP = METADATA$START_SCAN_TIME
    $$,
    P_ON_ERROR => 'SKIP_FILE'
);

{"copy_query_id":"01c79bb7-0001-7e01-0001-48f20009d34a","errors_seen":2,"load_failed_file_count":1,"load_status":"ERROR","loaded_file_count":0,"partially_loaded_file_count":0,"processed_file_count":1,"rows_loaded":0,"rows_parsed":0}

raise_error パラメータを TRUE にして処理結果を確認し、エラーとなることを確認します。

CALL SP_CHECK_COPY_RESULT(
    '{"copy_query_id":"01c79bb7-0001-7e01-0001-48f20009d34a","errors_seen":2,"load_failed_file_count":1,"load_status":"ERROR","loaded_file_count":0,"partially_loaded_file_count":0,"processed_file_count":1,"rows_loaded":0,"rows_parsed":0}', -- ToDo SP_COPY_ORDERS のリターン結果に書き換える
    TRUE
);

Uncaught exception of type 'COPY_LOAD_ERROR' on line 38 at position 12 : COPY INTO detected LOAD_FAILED files.

raise_error パラメータを FALSE にして処理結果を確認し、エラーとはならずにメッセージのみが返されることを確認します。

CALL SP_CHECK_COPY_RESULT(
    '{"copy_query_id":"01c77fd9-0001-6c72-0001-48f20004b93a","errors_seen":2,"load_failed_file_count":1,"load_status":"ERROR","loaded_file_count":0,"partially_loaded_file_count":0,"processed_file_count":1,"rows_loaded":0,"rows_parsed":0}', -- ToDo SP_COPY_ORDERS のリターン結果に書き換える
    FALSE
);

COPY INTO completed with LOAD_FAILED files. Load failed file count: 1. Query ID: 01c77fd9-0001-6c72-0001-48f20004b93a

過去データの削除

検証用のデータを挿入します。

INSERT INTO ORDERS (
    ORDER_ID,
    CUSTOMER_NAME,
    AMOUNT,
    ORDER_DATE,
    AUDIT__INGEST_TIMESTAMP
)
VALUES
    (
        '9001',
        'OLD_CUSTOMER',
        '100',
        '2026-01-01',
        DATEADD(
            DAY,
            -61,
            CURRENT_TIMESTAMP()
        )
    ),
    (
        '9002',
        'NEW_CUSTOMER',
        '200',
        '2026-10-04',
        DATEADD(
            DAY,
            -59,
            CURRENT_TIMESTAMP()
        )
    );

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

CALL SP_DELETE_OLD_ORDERS();

Deleted 1 rows older than 60 days.

9001 のレコードが削除され、9002 のレコードのみが表示されることを確認します。

SELECT
    ORDER_ID,
    AUDIT__INGEST_TIMESTAMP
FROM ORDERS
WHERE
    ORDER_ID IN (
        '9001',
        '9002'
    );

一連の処理として実行

TASK を定義

CREATE OR REPLACE TASK COPY_TEST_DB.TEST_SCHEMA.TASK_COPY_ORDERS
    USER_TASK_MANAGED_INITIAL_WAREHOUSE_SIZE = 'XSMALL'
AS
DECLARE
    copy_result     VARCHAR;
    check_result    VARCHAR;
    delete_result   VARCHAR;
    retention_days  NUMBER DEFAULT 0;
    raise_load_failed_error BOOLEAN DEFAULT TRUE;
    raise_partially_loaded_error BOOLEAN DEFAULT FALSE;

BEGIN
    -- =========================================================
    -- 0. 監査日時を UTC 基準で保存するためのセッション設定
    -- =========================================================
    -- METADATA$FILE_LAST_MODIFIED は TIMESTAMP_NTZ、
    -- METADATA$START_SCAN_TIME は TIMESTAMP_LTZ として返される。
    -- START_SCAN_TIME を監査列の TIMESTAMP_NTZ に変換する際の
    -- タイムゾーンを UTC に固定する。
    ALTER SESSION SET TIMEZONE = 'UTC';

    -- =========================================================
    -- 1. COPY INTO を実行
    -- =========================================================
    CALL COPY_TEST_DB.TEST_SCHEMA.SP_COPY(
        P_TARGET_TABLE => 'COPY_TEST_DB.TEST_SCHEMA.ORDERS',
        P_SOURCE_STAGE => 'COPY_TEST_DB.TEST_SCHEMA.INPUT_STAGE',
        P_SOURCE_PATH  => 'test_files/',
        P_FILE_FORMAT => $$
            TYPE = CSV
            PARSE_HEADER = TRUE
            FIELD_OPTIONALLY_ENCLOSED_BY = '"'
            ERROR_ON_COLUMN_COUNT_MISMATCH = FALSE
        $$,
        P_MATCH_BY_COLUMN_NAME => 'CASE_INSENSITIVE',
        P_INCLUDE_METADATA => $$
            AUDIT__INGEST_TIMESTAMP = METADATA$FILE_LAST_MODIFIED,
            AUDIT__UPDATE_TIMESTAMP = METADATA$START_SCAN_TIME
        $$,
        P_ON_ERROR => 'SKIP_FILE'
    )
        INTO :copy_result;

    -- =========================================================
    -- 2. COPY INTO の実行結果をすぐに Task の戻り値として設定
    -- =========================================================
    CALL SYSTEM$SET_RETURN_VALUE(:copy_result);

    -- =========================================================
    -- 3. COPY INTO の実行結果を判定
    -- =========================================================
    CALL COPY_TEST_DB.TEST_SCHEMA.SP_CHECK_COPY_RESULT(
        :copy_result,
        :raise_load_failed_error,
        :raise_partially_loaded_error
    )
        INTO :check_result;

    -- =========================================================
    -- 4. 古いデータを削除
    --    保持日数が 0 以下の場合は実行しない
    -- =========================================================
    IF (retention_days > 0) THEN
        CALL COPY_TEST_DB.TEST_SCHEMA.SP_DELETE_OLD_ORDERS(
            :retention_days
        )
            INTO :delete_result;
    END IF;
END;

CREATE OR REPLACE TASK TASK_DUMMY
    USER_TASK_MANAGED_INITIAL_WAREHOUSE_SIZE = 'XSMALL'
    AFTER COPY_TEST_DB.TEST_SCHEMA.TASK_COPY_ORDERS
AS
BEGIN
    SELECT 1;
END;

TASK_DUMMYタスクは TASK の処理をグラフにするために設定しております。

-- 1. 子 Task を有効化
ALTER TASK TASK_DUMMY RESUME;

正常ファイル取り込み時の一連の処理

ファイルを配置します。

normal_csv = """order_id,customer_name,amount,order_date
1,Alice,100.50,2026-10-04
2,Bob,200.00,2026-10-04
"""

reset_test_files()

put_csv(
    "normal_001.csv",
    normal_csv
)

一連の処理を実行します。

EXECUTE TASK COPY_TEST_DB.TEST_SCHEMA.TASK_COPY_ORDERS;

カタログエクスプローラーにて実行結果を確認します。

異常ファイル取り込み時の一連の処理

ファイルを配置します。

invalid_csv = """order_id,customer_name,amount,order_date
3,Carol,300.00,2026-10-04
4,David,400.00,2026-10-04,EXTRA_COLUMN
"""

reset_test_files()

put_csv(
    "invalid_001.csv",
    invalid_csv
)

一連の処理を実行します。

EXECUTE TASK COPY_TEST_DB.TEST_SCHEMA.TASK_COPY_ORDERS;

カタログエクスプローラーにて実行結果を確認します。

You get articles that match your needs
You can efficiently read back useful information
You can use dark theme

What you can do with signing up