如何用Snowflake JavaScript存储过程实现SCD Type 2及数组插入问题解决

一、核心问题解决：字符串转ARRAY插入失败 #

你的代码通过JS拼接SQL插入ARRAY的方式容易触发语法错误（比如子顾问名称含单引号时），且循环插入效率极低。以下两种方案可解决该问题：

方案1：用Snowflake内置函数直接生成ARRAY（最优） #

完全无需JS循环处理，直接在创建临时表时用SPLIT_TO_ARRAY函数将拼接后的字符串转为ARRAY类型，一步到位：

CREATE OR REPLACE PROCEDURE SUBADVISOR_CHANGES()
RETURNS STRING
LANGUAGE JAVASCRIPT
AS
$$
try {
    // 清理临时表
    snowflake.execute({sqlText: `DROP TABLE IF EXISTS DATAHUB_CORE_DEV_DB.SIL_STG_SCHEMA.CONCATSUBADVISORITER;`});

    // 直接生成带ARRAY类型的临时表，跳过中间字符串拼接环节
    snowflake.execute({sqlText: `
        CREATE TEMPORARY TABLE DATAHUB_CORE_DEV_DB.SIL_STG_SCHEMA.CONCATSUBADVISORITER AS (
            SELECT
                CUSIP,
                SPLIT_TO_ARRAY(LISTAGG(SUB_ADVISOR, ','), ',') AS CONCAT_SUBADVISOR,
                TO_DATE(JOB_RUN_DATE, 'YYYY-MM-DD') AS JOB_RUN_DATE
            FROM DATAHUB_STG_DEV_DB.FTP_STG_SCHEMA.FP_SUB_ADVISOR_STAGING
            GROUP BY CUSIP, JOB_RUN_DATE
        );
    `});

    return "Success";
} catch (e) {
    return "Error: " + e.message;
}
$$;

说明：

• SPLIT_TO_ARRAY直接将LISTAGG拼接的字符串按逗号分割为ARRAY类型，无需JS额外处理
• 跳过中间临时表CONCATSUBADVISOR，减少IO操作，提升性能
• 新增异常捕获逻辑，方便排查执行错误

方案2：修复JS存储过程的插入逻辑（保留JS处理场景） #

若必须用JS拆分字符串，不要直接拼接SQL，改用绑定变量传递数组，避免语法错误和SQL注入风险：

CREATE OR REPLACE PROCEDURE SUBADVISOR_CHANGES()
RETURNS STRING
LANGUAGE JAVASCRIPT
AS
$$
try {
    // 清理临时表
    snowflake.execute({sqlText: `DROP TABLE IF EXISTS DATAHUB_CORE_DEV_DB.SIL_STG_SCHEMA.CONCATSUBADVISOR;`});
    snowflake.execute({sqlText: `DROP TABLE IF EXISTS DATAHUB_CORE_DEV_DB.SIL_STG_SCHEMA.CONCATSUBADVISORITER;`});

    // 创建字符串拼接的临时表
    snowflake.execute({sqlText: `
        CREATE TEMPORARY TABLE DATAHUB_CORE_DEV_DB.SIL_STG_SCHEMA.CONCATSUBADVISOR AS (
            SELECT
                CUSIP,
                LISTAGG(SUB_ADVISOR, ',') AS CONCAT_SUBADVISOR,
                TO_VARCHAR(JOB_RUN_DATE, 'YYYY-MM-DD') AS JOB_RUN_DATE
            FROM DATAHUB_STG_DEV_DB.FTP_STG_SCHEMA.FP_SUB_ADVISOR_STAGING
            GROUP BY CUSIP, JOB_RUN_DATE
        );
    `});

    // 创建目标临时表
    snowflake.execute({sqlText: `
        CREATE TEMPORARY TABLE DATAHUB_CORE_DEV_DB.SIL_STG_SCHEMA.CONCATSUBADVISORITER (
            CUSIP VARCHAR(255),
            CONCAT_SUBADVISOR ARRAY,
            JOB_RUN_DATE DATE
        )
    `});

    // 查询数据并插入，使用绑定变量规避SQL拼接错误
    var result = snowflake.execute({sqlText: `
        SELECT CUSIP, CONCAT_SUBADVISOR, TO_DATE(JOB_RUN_DATE, 'YYYY-MM-DD') AS JOB_RUN_DATE
        FROM DATAHUB_CORE_DEV_DB.SIL_STG_SCHEMA.CONCATSUBADVISOR
    `});

    while (result.next()) {
        var cusip = result.getColumnValue(1);
        var subadvisorStr = result.getColumnValue(2);
        var runDate = result.getColumnValue(3);
        // 拆分字符串为JS数组
        var subadvisorArray = subadvisorStr.split(',').map(s => s.trim());

        // 使用绑定变量执行插入，通过PARSE_JSON将JS数组转为Snowflake ARRAY
        snowflake.execute({
            sqlText: `INSERT INTO DATAHUB_CORE_DEV_DB.SIL_STG_SCHEMA.CONCATSUBADVISORITER VALUES (?, PARSE_JSON(?), ?)`,
            binds: [cusip, JSON.stringify(subadvisorArray), runDate]
        });
    }

    return "Success";
} catch (e) {
    return "Error: " + e.message;
}
$$;

说明：

• 用PARSE_JSON将JS数组的JSON字符串转为Snowflake ARRAY类型
• 使用绑定变量?传递参数，彻底避免字符串拼接导致的语法错误（比如子顾问名称含单引号）
• 新增异常捕获，便于调试执行问题

二、SCD Type 2表实现建议（仅保留当日+前日数据） #

结合你的需求，构建SCD Type 2表时，可通过MERGE语句对比当日和前日数据，处理所有者新增，并定期清理超期数据：

1. 创建SCD Type 2目标表 #

CREATE OR REPLACE TABLE DATAHUB_CORE_DEV_DB.SIL_STG_SCHEMA.SUBADVISOR_SCD2 (
    CUSIP VARCHAR(255),
    CONCAT_SUBADVISOR ARRAY,
    EFFECTIVE_DATE DATE,
    EXPIRY_DATE DATE,
    IS_CURRENT BOOLEAN
);

2. 在存储过程中新增SCD2处理逻辑 #

在临时表生成后，添加以下MERGE和清理逻辑：

-- MERGE到SCD2表，处理变更与新增
snowflake.execute({sqlText: `
    MERGE INTO DATAHUB_CORE_DEV_DB.SIL_STG_SCHEMA.SUBADVISOR_SCD2 tgt
    USING DATAHUB_CORE_DEV_DB.SIL_STG_SCHEMA.CONCATSUBADVISORITER src
        ON tgt.CUSIP = src.CUSIP AND tgt.IS_CURRENT = TRUE
    WHEN MATCHED AND tgt.CONCAT_SUBADVISOR != src.CONCAT_SUBADVISOR THEN
        UPDATE SET tgt.EXPIRY_DATE = src.JOB_RUN_DATE - 1, tgt.IS_CURRENT = FALSE
        INSERT (CUSIP, CONCAT_SUBADVISOR, EFFECTIVE_DATE, EXPIRY_DATE, IS_CURRENT)
        VALUES (src.CUSIP, src.CONCAT_SUBADVISOR, src.JOB_RUN_DATE, '9999-12-31', TRUE)
    WHEN NOT MATCHED THEN
        INSERT (CUSIP, CONCAT_SUBADVISOR, EFFECTIVE_DATE, EXPIRY_DATE, IS_CURRENT)
        VALUES (src.CUSIP, src.CONCAT_SUBADVISOR, src.JOB_RUN_DATE, '9999-12-31', TRUE);
`});

-- 清理早于前日的历史数据，确保仅保留当日和前日数据
snowflake.execute({sqlText: `
    DELETE FROM DATAHUB_CORE_DEV_DB.SIL_STG_SCHEMA.SUBADVISOR_SCD2
    WHERE EXPIRY_DATE < CURRENT_DATE() - 1;
`});

说明：

• MERGE语句处理SCD2变更：当CUSIP匹配但子顾问数组变化时，过期旧记录并插入新记录；未匹配的新增记录直接插入
• DELETE语句清理超期数据，严格控制表中仅保留当日和前日的有效/过期记录

内容的提问来源于stack exchange，提问作者pscodes