Snowflake存储过程报错：source_database未定义求助

云原生阿岚

2026-6-13

Snowflake存储过程报错：source_database is not defined

我编写了一个Snowflake存储过程，用于动态对比不同数据库、Schema下源表与目标表的行数，遍历源Schema中的所有表并记录行数差异。但执行该过程时抛出错误：❌ Procedure failed: source_database is not defined，报错截图如下：

执行存储过程时的报错截图

原代码如下：

CREATE OR REPLACE PROCEDURE TEST1_validate_all_table_row_counts(
    source_database STRING,
    source_schema STRING,
    target_database STRING,
    target_schema STRING
)
RETURNS STRING
LANGUAGE JAVASCRIPT
AS
$$
try {
    var result_log = [];

    // Convert input parameters to uppercase for consistency
    var src_db = source_database.toUpperCase();
    var src_schema = source_schema.toUpperCase();
    var tgt_db = target_database.toUpperCase();
    var tgt_schema = target_schema.toUpperCase();

    result_log.push(`🔹 Validating row counts between ${src_db}.${src_schema} and ${tgt_db}.${tgt_schema}`);

    // Get all tables from the source schema
    var sql_get_tables = `
        SELECT TABLE_NAME 
        FROM "` + src_db + `"."INFORMATION_SCHEMA"."TABLES"
        WHERE UPPER(TABLE_SCHEMA) = '` + src_schema + `'
        AND TABLE_TYPE = 'BASE TABLE'
        ORDER BY TABLE_NAME`;

    var stmt = snowflake.execute({ sqlText: sql_get_tables });

    var table_count = 0;
    while (stmt.next()) {
        table_count++;
        var table_name = stmt.getColumnValue(1);
        result_log.push(`🔹 Checking table: ${table_name}`);

        // Check if table exists in target schema
        var sql_check_target = `
            SELECT COUNT(*) 
            FROM "` + tgt_db + `"."INFORMATION_SCHEMA"."TABLES"
            WHERE UPPER(TABLE_SCHEMA) = '` + tgt_schema + `'
            AND UPPER(TABLE_NAME) = '` + table_name.toUpperCase() + `'`;

        var check_stmt = snowflake.execute({ sqlText: sql_check_target });
        check_stmt.next();
        var table_exists = check_stmt.getColumnValue(1);

        if (table_exists === 0) {
            result_log.push(`⚠️ Skipped: ${table_name} does not exist in target schema.`);
            continue;
        }

        // Row count queries
        var sql_source = `SELECT COUNT(*) FROM "` + src_db + `"."` + src_schema + `"."` + table_name + `"`;
        var sql_target = `SELECT COUNT(*) FROM "` + tgt_db + `"."` + tgt_schema + `"."` + table_name + `"`;

        result_log.push(`Executing SQL on Source: ${sql_source}`);
        result_log.push(`Executing SQL on Target: ${sql_target}`);

        try {
            var source_stmt = snowflake.execute({ sqlText: sql_source });
            var target_stmt = snowflake.execute({ sqlText: sql_target });

            source_stmt.next();
            target_stmt.next();

            var source_count = source_stmt.getColumnValue(1);
            var target_count = target_stmt.getColumnValue(1);

            if (source_count === target_count) {
                result_log.push(`✅ Match: ${table_name} -> Source: ${source_count}, Target: ${target_count}`);
            } else {
                result_log.push(`❌ Mismatch: ${table_name} -> Source: ${source_count}, Target: ${target_count}`);
            }
        } catch (err) {
            result_log.push(`⚠️ Error processing table ${table_name}: ${err.message}`);
        }
    }

    // If no tables were found, return a message
    if (table_count === 0) {
        result_log.push(`⚠️ No tables found in source schema: ${src_schema}`);
    }

    return result_log.join("\n");

} catch (err) {
    return `❌ Procedure failed: ${err.message}`;
}
$$;

问题原因

Snowflake的JavaScript存储过程不支持直接通过参数名访问输入参数，必须使用位置引用（$1、$2...）或arguments数组来获取参数值。原代码直接使用source_database等参数名，导致变量未定义报错。

修复方案

方案1：使用arguments数组获取参数

在代码开头先通过arguments数组获取所有参数值，再进行后续处理：

CREATE OR REPLACE PROCEDURE TEST1_validate_all_table_row_counts(
    source_database STRING,
    source_schema STRING,
    target_database STRING,
    target_schema STRING
)
RETURNS STRING
LANGUAGE JAVASCRIPT
AS
$$
try {
    var result_log = [];

    // 从arguments数组获取输入参数
    var source_database = arguments[0];
    var source_schema = arguments[1];
    var target_database = arguments[2];
    var target_schema = arguments[3];

    // Convert input parameters to uppercase for consistency
    var src_db = source_database.toUpperCase();
    var src_schema = source_schema.toUpperCase();
    var tgt_db = target_database.toUpperCase();
    var tgt_schema = target_schema.toUpperCase();

    result_log.push(`🔹 Validating row counts between ${src_db}.${src_schema} and ${tgt_db}.${tgt_schema}`);

    var sql_get_tables = `
        SELECT TABLE_NAME 
        FROM "` + src_db + `"."INFORMATION_SCHEMA"."TABLES"
        WHERE UPPER(TABLE_SCHEMA) = '` + src_schema + `'
        AND TABLE_TYPE = 'BASE TABLE'
        ORDER BY TABLE_NAME`;

    var stmt = snowflake.execute({ sqlText: sql_get_tables });

    var table_count = 0;
    while (stmt.next()) {
        table_count++;
        var table_name = stmt.getColumnValue(1);
        result_log.push(`🔹 Checking table: ${table_name}`);

        var sql_check_target = `
            SELECT COUNT(*) 
            FROM "` + tgt_db + `"."INFORMATION_SCHEMA"."TABLES"
            WHERE UPPER(TABLE_SCHEMA) = '` + tgt_schema + `'
            AND UPPER(TABLE_NAME) = '` + table_name.toUpperCase() + `'`;

        var check_stmt = snowflake.execute({ sqlText: sql_check_target });
        check_stmt.next();
        var table_exists = check_stmt.getColumnValue(1);

        if (table_exists === 0) {
            result_log.push(`⚠️ Skipped: ${table_name} does not exist in target schema.`);
            continue;
        }

        var sql_source = `SELECT COUNT(*) FROM "` + src_db + `"."` + src_schema + `"."` + table_name + `"`;
        var sql_target = `SELECT COUNT(*) FROM "` + tgt_db + `"."` + tgt_schema + `"."` + table_name + `"`;

        result_log.push(`Executing SQL on Source: ${sql_source}`);
        result_log.push(`Executing SQL on Target: ${sql_target}`);

        try {
            var source_stmt = snowflake.execute({ sqlText: sql_source });
            var target_stmt = snowflake.execute({ sqlText: sql_target });

            source_stmt.next();
            target_stmt.next();

            var source_count = source_stmt.getColumnValue(1);
            var target_count = target_stmt.getColumnValue(1);

            if (source_count === target_count) {
                result_log.push(`✅ Match: ${table_name} -> Source: ${source_count}, Target: ${target_count}`);
            } else {
                result_log.push(`❌ Mismatch: ${table_name} -> Source: ${source_count}, Target: ${target_count}`);
            }
        } catch (err) {
            result_log.push(`⚠️ Error processing table ${table_name}: ${err.message}`);
        }
    }

    if (table_count === 0) {
        result_log.push(`⚠️ No tables found in source schema: ${src_schema}`);
    }

    return result_log.join("\n");

} catch (err) {
    return `❌ Procedure failed: ${err.message}`;
}
$$;

方案2：直接使用位置参数$1-$4（更简洁）

直接用$1对应第一个参数、$2对应第二个参数，以此类推，无需额外定义参数变量：

CREATE OR REPLACE PROCEDURE TEST1_validate_all_table_row_counts(
    source_database STRING,
    source_schema STRING,
    target_database STRING,
    target_schema STRING
)
RETURNS STRING
LANGUAGE JAVASCRIPT
AS
$$
try {
    var result_log = [];

    // 直接使用位置参数$1-$4转换为大写
    var src_db = $1.toUpperCase();
    var src_schema = $2.toUpperCase();
    var tgt_db = $3.toUpperCase();
    var tgt_schema = $4.toUpperCase();

    result_log.push(`🔹 Validating row counts between ${src_db}.${src_schema} and ${tgt_db}.${tgt_schema}`);

    var sql_get_tables = `
        SELECT TABLE_NAME 
        FROM "` + src_db + `"."INFORMATION_SCHEMA"."TABLES"
        WHERE UPPER(TABLE_SCHEMA) = '` + src_schema + `'
        AND TABLE_TYPE = 'BASE TABLE'
        ORDER BY TABLE_NAME`;

    var stmt = snowflake.execute({ sqlText: sql_get_tables });

    var table_count = 0;
    while (stmt.next()) {
        table_count++;
        var table_name = stmt.getColumnValue(1);
        result_log.push(`🔹 Checking table: ${table_name}`);

        var sql_check_target = `
            SELECT COUNT(*) 
            FROM "` + tgt_db + `"."INFORMATION_SCHEMA"."TABLES"
            WHERE UPPER(TABLE_SCHEMA) = '` + tgt_schema + `'
            AND UPPER(TABLE_NAME) = '` + table_name.toUpperCase() + `'`;

        var check_stmt = snowflake.execute({ sqlText: sql_check_target });
        check_stmt.next();
        var table_exists = check_stmt.getColumnValue(1);

        if (table_exists === 0) {
            result_log.push(`⚠️ Skipped: ${table_name} does not exist in target schema.`);
            continue;
        }

        var sql_source = `SELECT COUNT(*) FROM "` + src_db + `"."` + src_schema + `"."` + table_name + `"`;
        var sql_target = `SELECT COUNT(*) FROM "` + tgt_db + `"."` + tgt_schema + `"."` + table_name + `"`;

        result_log.push(`Executing SQL on Source: ${sql_source}`);
        result_log.push(`Executing SQL on Target: ${sql_target}`);

        try {
            var source_stmt = snowflake.execute({ sqlText: sql_source });
            var target_stmt = snowflake.execute({ sqlText: sql_target });

            source_stmt.next();
            target_stmt.next();

            var source_count = source_stmt.getColumnValue(1);
            var target_count = target_stmt.getColumnValue(1);

            if (source_count === target_count) {
                result_log.push(`✅ Match: ${table_name} -> Source: ${source_count}, Target: ${target_count}`);
            } else {
                result_log.push(`❌ Mismatch: ${table_name} -> Source: ${source_count}, Target: ${target_count}`);
            }
        } catch (err) {
            result_log.push(`⚠️ Error processing table ${table_name}: ${err.message}`);
        }
    }

    if (table_count === 0) {
        result_log.push(`⚠️ No tables found in source schema: ${src_schema}`);
    }

    return result_log.join("\n");

} catch (err) {
    return `❌ Procedure failed: ${err.message}`;
}
$$;

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

Snowflake存储过程报错：source_database未定义求助

问题原因 #

修复方案 #

方案1：使用arguments数组获取参数 #

方案2：直接使用位置参数$1-$4（更简洁） #

问题原因

修复方案

方案1：使用arguments数组获取参数

方案2：直接使用位置参数$1-$4（更简洁）