如何修改Oracle存储过程按每千条记录生成CSV压缩文件

修改核心逻辑 #

• 为查询结果添加行号，通过行号计算批次分组（每1000条为一个批次）
• 新增批次编号、当前批次行数字段，实时跟踪处理进度
• 每累计1000条记录时，立即生成压缩文件并插入数据库，随后重置临时存储变量
• 循环结束后处理剩余的不足1000条记录，生成最后一个压缩包
• 调整文件名规则，加入批次号确保每个压缩文件唯一

改造后的完整存储过程代码 #

CREATE OR REPLACE PROCEDURE generate_fbdi_sp (
        p_id_proceso_oic  IN VARCHAR2,
        p_source          IN VARCHAR2,
        p_file_reference  IN VARCHAR2,
        out_result        OUT VARCHAR2,
        out_message       OUT VARCHAR2
    ) IS
    -- 带行号的游标，用于按批次划分记录
        CURSOR c_csv_fbdi IS
        SELECT
            status_code
            || ','
            || ledger_id
            || ','
            || effective_date_of_transaction
            || ','
            || journal_source
            || ','
            || journal_category
            || ','
            || currency_code
            || ','
            || journal_entry_creation_date
            || CHR(10) AS linea,
            -- 添加行号，用于计算批次
            ROWNUM AS row_num
        FROM
            int_dat_journals_fbdi_layout
        WHERE
                journal_source = p_source
            AND id_proceso_oic = p_id_proceso_oic;
            
        -- 变量声明
        v_csv_line      CLOB;
        g_zipped_blob   BLOB;
        v_blob_csv_line BLOB;
        v_row_count     NUMBER := 0; -- 当前批次已处理行数
        v_batch_number  NUMBER := 1; -- 当前批次编号
    BEGIN
        -- 初始化临时LOB变量
        DBMS_LOB.CREATETEMPORARY(v_csv_line, TRUE);
        DBMS_LOB.CREATETEMPORARY(g_zipped_blob, TRUE);
        
        -- 逐行处理记录
        FOR curr_line IN c_csv_fbdi LOOP
            v_row_count := v_row_count + 1;
            -- 将当前行追加到CLOB
            DBMS_LOB.WRITEAPPEND(v_csv_line, LENGTH(curr_line.linea), curr_line.linea);
            
            -- 每满1000条，生成压缩包并存储
            IF v_row_count = 1000 THEN
                -- CLOB转BLOB
                SELECT clob_to_blob_fn(v_csv_line) INTO v_blob_csv_line FROM dual;
                
                -- 生成压缩文件
                as_zip.add1file(g_zipped_blob, 'GlInterface_'
                                       || replace(p_source, ' ', '_')
                                       || '_'
                                       || p_id_proceso_oic
                                       || '_batch_' || v_batch_number
                                       || '.csv', v_blob_csv_line);
                as_zip.finish_zip(g_zipped_blob);
                
                -- 插入数据库
                INSERT INTO journals_fbdi (
                    file_name,
                    file_content,
                    id_proceso_oic,
                    file_name_cv027,
                    source
                ) VALUES (
                    'FBDI_'
                    || replace(p_source, ' ', '_')
                    || '_'
                    || p_id_proceso_oic
                    || '_batch_' || v_batch_number
                    || '.zip',
                    g_zipped_blob,
                    p_id_proceso_oic,
                    p_file_reference,
                    p_source
                );
                
                -- 重置变量，准备下一批次
                DBMS_LOB.TRIM(v_csv_line, 0);
                DBMS_LOB.TRIM(g_zipped_blob, 0);
                v_row_count := 0;
                v_batch_number := v_batch_number + 1;
            END IF;
        END LOOP;
        
        -- 处理最后一批不足1000条的记录
        IF v_row_count > 0 THEN
            SELECT clob_to_blob_fn(v_csv_line) INTO v_blob_csv_line FROM dual;
            
            as_zip.add1file(g_zipped_blob, 'GlInterface_'
                                   || replace(p_source, ' ', '_')
                                   || '_'
                                   || p_id_proceso_oic
                                   || '_batch_' || v_batch_number
                                   || '.csv', v_blob_csv_line);
            as_zip.finish_zip(g_zipped_blob);
            
            INSERT INTO journals_fbdi (
                file_name,
                file_content,
                id_proceso_oic,
                file_name_cv027,
                source
            ) VALUES (
                'FBDI_'
                || replace(p_source, ' ', '_')
                || '_'
                || p_id_proceso_oic
                || '_batch_' || v_batch_number
                || '.zip',
                g_zipped_blob,
                p_id_proceso_oic,
                p_file_reference,
                p_source
            );
        END IF;
        
        COMMIT;
        out_result := 'SUCCESS';
        out_message := '所有批次文件已生成并存储完成';
        
    EXCEPTION
        WHEN OTHERS THEN
            ROLLBACK;
            out_result := 'ERROR: ' || sqlerrm;
            out_message := dbms_utility.format_error_backtrace;
    END;

关键修改说明 #

1. 游标添加行号：通过ROWNUM为每条记录标记行号，用于判断批次划分
2. 批次跟踪变量：新增v_row_count记录当前批次行数，v_batch_number标记当前批次号
3. 批量处理逻辑：每累计1000条记录时，立即执行压缩、存储操作，并重置临时变量
4. 剩余记录处理：循环结束后检查是否有未处理的剩余记录，确保所有数据都被生成文件
5. 文件名唯一化：在文件名中加入_batch_<批次号>，避免不同批次文件重名
6. LOB变量优化：使用DBMS_LOB的CREATETEMPORARY和TRIM方法管理CLOB/BLOB变量，避免内存溢出

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