如何解决执行存储过程导致的内存不足错误？

Posted 2023-03-29

【中文标题】如何解决执行存储过程导致的内存不足错误？

【英文标题】：How to fix out of memory error from executing a stored procedure?

【发布时间】：2020-09-30 12:48:38

【问题描述】：

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

数据最初加载到暂存数据库中的表中。从暂存表中，我在将主表与临时表交换之前将数据加载到临时表中，这可以在方法中看到：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 错误 [100176] [P0000]：JavaScript 内存不足错误：超出 UDF 线程内存限制

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

原因：net.snowflake.client.jdbc.SnowflakeSQLException: JavaScript out of memory error: UDF thread memory limit exceeded

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

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

这是我第一次处理存储过程和 java 脚本。我不明白是什么导致了这个错误以及如何分析这个错误。谁能让我知道如何解决这个问题？非常感谢任何帮助。

【参考方案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

【参考方案2】：

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

【讨论】：

能否请您告诉我们，您是如何调试的..？你有没有在 try catch 块中发现它......怎么......？

是的，我已将 while 循环放在 TRY-CATCH 块中，并在 CATCH 中打印了日志消息。那里的代码被破坏了。

你是如何通过 console.log() 打印的？

哦，不。我正在从存储过程中返回一个 Json 对象。此 Json 包含来自所有 try 和 catch 块的消息（和错误）字符串。它是这样的：apptrace = "PRIMARY KEYS": pk, "UNIQUE KEYS": uk, "MESSAGE": logs "TS": datetime ;return JSON.stringify(apptrace); => 我在 dbeaver 上运行 storeproc。我对 Javascript 很陌生。我知道我们可以使用 console.log() 但不知道如何在像 Dbeaver 这样的数据库工具上使用它。所以努力返回一个 JSON 字符串。