使用 Snowflake 存储过程将文件从 AWS S3 复制到 Snowflake 表

Question

当我尝试使用存储过程动态执行将文件从 S3 存储桶复制到 Snowflake 表的过程时，它抛出以下错误：

** SQL编译错误： 未知函数 UDF_GET_COPY_COLUMNS 在 Statement.execute，第 4 行位置 20**

CREATE OR REPLACE PROCEDURE SF_COPY_LEAD_COLLECT_STAGE()
RETURNS VARIANT NOT NULL
LANGUAGE JAVASCRIPT
AS
$$
var row_as_json = {};
var udfstmt = snowflake.createStatement({sqlText: "SELECT UDF_GET_COPY_COLUMNS();"})
var udfrs = udfstmt.execute();
udfrs.next();
var COL_NAMES = udfrs.getColumnValue(1);
var command = "COPY INTO SF_ADVEN_COLLECT_DB.PRODUCT_HUB.PRODUCT FROM @DEV_NP_ADV_RAW_DB.PRODUCT_HUBS.NP_ADVEN_EXT_STG/PRODUCT/PRODUCT_202401121_231143.txt; ";
var stmt = snowflake.createStatement({sqlText: command});
var rs = stmt.execute();
//Move to the First Row Returned 
rs.next();
//check for error 
var errors_seen = rs.getColumnValue(6);
if (errors_seen > 0) 
{
throw new Error("Copy Command encountered Errors: "+errors_seen);
}
// Loop through the columns retured by the copy command.push each as key-value pair into a json object to return
for (var col_num = 0; col_num < COL_NAMES.length; col_num = col_num + 1)
{
var col_name = COL_NAMES[col_num];
row_as_json[col_name] = rs.getColumnValue(col_num + 1);
}
return row_as_json;
$

$

Answer 1

可能有几种可能的原因导致函数 UDF_GET_COPY_COLUMNS() 出错

函数范围：函数 UDF_GET_COPY_COLUMNS() 可能在不同的模式中定义，并且您没有使用适当的模式名称引用它。

上下文问题： 如果函数是在不同的角色中定义的，或者是不同类型的对象（例如，存储过程或视图），Snowflake 可能无法正确解析该函数。

您可以检查几件事：

检查函数是否存在： 运行以下查询来检查当前数据库/模式中是否存在函数 UDF_GET_COPY_COLUMNS()：

SHOW FUNCTIONS LIKE 'UDF_GET_COPY_COLUMNS';

如果该函数不存在，您可能需要定义它或确保它在正确的架构中可用。如果确实存在，请确保架构正确。

显式引用架构：如果函数位于不同的架构中，请确保通过其架构名称显式引用它。例如：

var udfstmt = snowflake.createStatement({sqlText: "SELECT <schema_name>.UDF_GET_COPY_COLUMNS();"});

检查访问权限：确保您使用的角色具有访问函数 UDF_GET_COPY_COLUMNS() 所需的权限。

替代选项：

如果您不需要 UDF 并且只想动态获取 COPY INTO 操作的列名称，您也可以查询要复制到的表的元数据。例如，您可以使用 INFORMATION_SCHEMA.COLUMNS 表动态检索列名：

var udfstmt = snowflake.createStatement({
    sqlText: `SELECT COLUMN_NAME 
              FROM INFORMATION_SCHEMA.COLUMNS 
              WHERE TABLE_NAME = 'PRODUCT' 
              AND TABLE_SCHEMA = 'PRODUCT_HUB' 
              AND TABLE_CATALOG = 'SF_ADVEN_COLLECT_DB'`
});

var udfrs = udfstmt.execute();
var COL_NAMES = [];
while (udfrs.next()) {
    COL_NAMES.push(udfrs.getColumnValue(1));
}