如何修复执行存储过程时出现的内存不足错误？

Question

我正在尝试执行以下雪花存储过程：

数据最初加载到 table 暂存数据库中。从暂存 table，我将数据加载到临时 table，然后将主 table 与临时 table 交换，这可以在方法中看到：temp_table_insert_sql

成功完成后，将在方法中执行交换：getSwapTableStmt。

下面是我要执行的存储过程。

CREATE OR REPLACE PROCEDURE "DBNAME"."SCHEMA"."FULL_LOAD_WITH_RETRY_FACILITY"(RETRY_CNT FLOAT, MIN_WAIT_SECOND FLOAT, MAX_WAIT_SECOND FLOAT, TABLE_NAME VARCHAR, TEMPDB VARCHAR, TEMPSCHEMA VARCHAR, TARGETDB VARCHAR, TARGETSCHEMA VARCHAR)
RETURNS VARCHAR(16777216)
LANGUAGE JAVASCRIPT
EXECUTE AS OWNER
AS $$

    function getRandom(min,max){
        return Math.floor(Math.random()*(max-min+1))+min;
    }

    function create_temp_table(table_name) {
        var create_temp_table = `CREATE OR REPLACE TABLE ${TEMPDB}.${TEMPSCHEMA}.${table_name}_TEMP like ${TARGETDB}.${TARGETSCHEMA}.${table_name};`;
        return create_temp_table;
    }

    function retry(MIN_WAIT_SECOND, MAX_WAIT_SECOND, retry_cnt, stmt) {
        for (cnt=0; cnt<retry_cnt; cnt++) {
            try {
                var result = stmt.execute();
                break;
            } catch (err) {
                var get_error_queryid_result = snowflake.createStatement( { sqlText:`SELECT LAST_QUERY_ID();` } ).execute();
                get_error_queryid_result.next();
                var queryId = String( get_error_queryid_result.getColumnValue(1) ).trim();
                snowflake.createStatement( { sqlText: `SELECT SYSTEM$WAIT(${getRandom(MIN_WAIT_SECOND,MAX_WAIT_SECOND)});` } ).execute();
                var err_msg = err.message;
                if(!(err_msg.includes("locked")) | (cnt==(retry_cnt-1)) ){
                    err.stackTraceTxt += ` QueryId: ${queryId}`;
                    throw err;
                }
            }
        }
        return result;
    }

    function temp_table_insert_sql(table_name) {
        var select_columns = "";
        var insert_columns = "";
        var sql = `select column_name from ${TARGETDB}.INFORMATION_SCHEMA.COLUMNS where true and table_schema = '${TARGETSCHEMA}' and table_name = '${table_name}' order by ordinal_position;`;
        var result = snowflake.createStatement( { sqlText: sql } ).execute();
        if (result.next()) {
            insert_columns = `( ${String(result.getColumnValue(1).trim)}`;
            select_columns = `select ${String(result.getColumnValue(1).trim)}`;
            while(result.next) {
                insert_columns += `, ${String(result.getColumnValue(1).trim)}`;
                select_columns += `, ${String(result.getColumnValue(1).trim)}`;
            }
            insert_columns += `)`;
        } else {
            insert_columns = "";
        }
        return `CREATE OR REPLACE TABLE ${TEMPDB}.${TEMPSCHEMA}.${table_name}_TEMP AS ${select_columns} from ${TEMPDB}.${TEMPSCHEMA}.${table_name}_STG`;
    }

    function get_row_count(db, SCHEMA, table_name) {
        return `select count(*) from ${db}.${schema}.${table_name};`;
    }

    function getSwapTableStmt(table_name){
        return `alter table if exists ${TARGETDB}.${TARGETSCHEMA}.${table_name} swap with ${TEMPDB}.${TEMPSCHEMA}.${table_name}_TEMP;`;
    }
    
    try {
        var timest = new Date().getTime();
        var temp_row_count = 0;
        var stg_row_count = 0;
        var apptrace = "";
        var logs = "";
        var result = "";
        var sql = "";

        var create_temp_ddl = create_temp_table(TABLE_NAME);
        logs = create_temp_ddl;
        result = retry(MIN_WAIT_SECOND, MAX_WAIT_SECOND, RETRY_CNT, snowflake.createStatement({sqlText: create_temp_ddl}));

        var insert_into_temp_table = temp_table_insert_sql(TABLE_NAME);
        result = retry(MIN_WAIT_SECOND, MAX_WAIT_SECOND, RETRY_CNT, snowflake.createStatement({ sqlText: insert_into_temp_table }));
        
        var temprows = get_row_count(TEMPDB, TABLE_NAME + "_TEMP");
        result = retry(MIN_WAIT_SECOND, MAX_WAIT_SECOND, RETRY_CNT, snowflake.createStatement({ sqlText: temprows }));
        if (result.next()) {
            temp_row_count;
        } else {
            var err = new Error("rows inserted is zero");
            throw err;
        }

        var stgrows = get_row_count(TEMPDB, TABLE_NAME + "_STG");
        result = retry(MIN_WAIT_SECOND, MAX_WAIT_SECOND, RETRY_CNT, snowflake.createStatement({ sqlText: stgrows }));
        if (result.next()) {
            stg_row_count = parseInt(result.getColumnValue(1));
        } else {
            var err = new Error("rows inserted is zero");
            throw err;
        }

        logs = `SWAP ${TABLE_NAME}_TEMP ${TABLE_NAME};`;
        if(stg_row_count == temp_row_count) {
            sql = getSwapTableStmt(table_name);
        } else {
            var err = new Error(`rows of ${TABLE_NAME}_STG & ${TABLE_NAME}_TEMP are different`);
            throw err;
        }
        var duration = (new Date().getTime() - timest) / 1000;
        apptrace = `{
            "application_name": "FULL_LOAD"
            ,"feature_name": "exchange|3.0|SESSION|FILE|${TABLE_NAME}_STG|${TABLE_NAME}"
            ,"event_subtype": "Metric"
            ,"metrics": [
                {
                    "metric":"Execution_Result"
                    ,"measurement":1
                    ,"unit_of_measure":"Boolean"
                }
                ,{
                    "metric": "Execution_Duration"
                    ,"measurement": ${duration}
                    ,"unit_of_measure": "Seconds"
                }
                ,{
                    "metric": "Rows"
                    ,"measurement": ${temp_row_count}
                    ,"unit_of_measure": "Rows"
                }
            ]
        }`;
    } catch (err)  {
    logs += "  Code: " + err.code + "  State: " + err.state;
    logs += "  Message: " + err.message;
    apptrace = `{
        "application_name": "FULL_LOAD"
        ,"feature_name": "exchange|3.0|SESSION|FILE|${TABLE_NAME}_STG|${TABLE_NAME}|${logs}"
        ,"event_subtype": "Metric"
        ,"metrics": [
            {
                "metric":"Execution_Result"
                ,"measurement":0
                ,"unit_of_measure":"Boolean"
            }
        ]
    }`;
    }
    // COMMIT
    snowflake.execute (
        {sqlText: `COMMIT;`}
    );
    return apptrace;
$$
;

这就是我调用存储过程的方式：

CALL DBNAME.SCHEMANAME.PROCNAME(2, 1000, 5000, 'TABLENAME', 'TEMPDB', 'TEMPSCHEMA', 'TARGETDB ', 'TARGETSCHEMA');

提交上述语句后，出现错误：

org.jkiss.dbeaver.model.sql.DBSQLException: SQL Error [100176] [P0000]: JavaScript out of memory error: UDF thread memory limit exceeded

at org.jkiss.dbeaver.model.impl.jdbc.exec.JDBCPreparedStatementImpl.executeStatement(JDBCPreparedStatementImpl.java:208)
at org.jkiss.dbeaver.ui.editors.sql.execute.SQLQueryJob.executeStatement(SQLQueryJob.java:492)
at org.jkiss.dbeaver.ui.editors.sql.execute.SQLQueryJob.lambda[=15=](SQLQueryJob.java:427)
at org.jkiss.dbeaver.model.exec.DBExecUtils.tryExecuteRecover(DBExecUtils.java:170)
at org.jkiss.dbeaver.ui.editors.sql.execute.SQLQueryJob.executeSingleQuery(SQLQueryJob.java:419)
at org.jkiss.dbeaver.ui.editors.sql.execute.SQLQueryJob.extractData(SQLQueryJob.java:779)
at org.jkiss.dbeaver.ui.editors.sql.SQLEditor$QueryResultsContainer.readData(SQLEditor.java:2973)
at org.jkiss.dbeaver.ui.controls.resultset.ResultSetJobDataRead.lambda[=15=](ResultSetJobDataRead.java:111)
at org.jkiss.dbeaver.model.exec.DBExecUtils.tryExecuteRecover(DBExecUtils.java:170)
at org.jkiss.dbeaver.ui.controls.resultset.ResultSetJobDataRead.run(ResultSetJobDataRead.java:109)
at org.jkiss.dbeaver.ui.controls.resultset.ResultSetViewer.run(ResultSetViewer.java:3584)
at org.jkiss.dbeaver.model.runtime.AbstractJob.run(AbstractJob.java:104)
at org.eclipse.core.internal.jobs.Worker.run(Worker.java:63)

Caused by: net.snowflake.client.jdbc.SnowflakeSQLException: JavaScript out of memory error: UDF thread memory limit exceeded

at net.snowflake.client.jdbc.SnowflakeUtil.checkErrorAndThrowExceptionSub(SnowflakeUtil.java:153)
at net.snowflake.client.jdbc.SnowflakeUtil.checkErrorAndThrowException(SnowflakeUtil.java:77)
at net.snowflake.client.core.StmtUtil.pollForOutput(StmtUtil.java:503)
at net.snowflake.client.core.StmtUtil.execute(StmtUtil.java:380)
at net.snowflake.client.core.SFStatement.executeHelper(SFStatement.java:582)
at net.snowflake.client.core.SFStatement.executeQueryInternal(SFStatement.java:266)
at net.snowflake.client.core.SFStatement.executeQuery(SFStatement.java:202)
at net.snowflake.client.core.SFStatement.execute(SFStatement.java:877)
at net.snowflake.client.jdbc.SnowflakeStatementV1.executeInternal(SnowflakeStatementV1.java:331)
at net.snowflake.client.jdbc.SnowflakePreparedStatementV1.execute(SnowflakePreparedStatementV1.java:535)
at org.jkiss.dbeaver.model.impl.jdbc.exec.JDBCPreparedStatementImpl.execute(JDBCPreparedStatementImpl.java:261)
at org.jkiss.dbeaver.model.impl.jdbc.exec.JDBCPreparedStatementImpl.executeStatement(JDBCPreparedStatementImpl.java:205)
... 12 more

数据库我是运行这个存储过程是Snowflake.

这是我第一次处理存储过程和 java 脚本。我不明白导致此错误的原因以及如何分析此错误。谁能告诉我如何解决这个问题？非常感谢任何帮助。

Answer 1

Snowflake 中 JavaScript UDFS 可以消耗的内存有限制： https://docs.snowflake.com/en/sql-reference/udf-js.html#memory

如果您尝试简化查询或通常将一笔交易拆分成几笔，则可能不会出现该错误。（这在一个案例中帮助了我）

也许这个参数对你有帮助：https://docs.snowflake.com/en/sql-reference/parameters.html#client-memory-limit

Answer 2

调试问题后，我发现代码中的while循环会无限循环并导致Out Of Memory Exception。在我修复了循环之后，错误就解决了。

如何修复执行存储过程时出现的内存不足错误？

How to fix out of memory error from executing a stored procedure?

mysql

stored-procedures

snowflake-cloud-data-platform