Oracle Array Processing with Bulk Collection & FORALL

Oracle Array Processing with Bulk Collection & FORALL Version 12.1.0.0.2

General Information

Without the bulk bind, PL/SQLsends a SQL statement to the SQL engine for each record that is inserted, updated, or deleted leading to context switches that hurt performance.

Syntax and Demo Directory

BULK COLLECT	FORALL INSERT	PARTIAL COLLECTIONS
Dynamic SQL Demos	FORALL MERGE	SAVE EXCEPTIONS
Exception Handling	FORALL UPDATE	SPARSE COLLECTIONS
FETCH FIRST	LIMIT CLAUSE	Performance Demos
FORALL DELETE	IN INDICES OF	Things to Consider

BULK COLLECT

BULK COLLECT Syntax and Demos

FETCH BULK COLLECT <cursor_name> BULK COLLECT INTO <collection_name>
[FETCH FIRST <integer> ROWS ONLY]
[LIMIT <numeric_expression>];

or

FETCH BULK COLLECT <cursor_name> BULK COLLECT INTO <array_name>
[FETCH FIRST <integer> ROWS ONLY]
[LIMIT <numeric_expression>];

set timing on

DECLARE
CURSOR a_cur IS
SELECT program_id
FROM airplanes;
BEGIN
FOR cur_rec IN a_cur LOOP
    NULL;
END LOOP;
END;
/

DECLARE
CURSOR a_cur IS
SELECT program_id
FROM airplanes;

TYPE myarray IS TABLE OF a_cur%ROWTYPE;
cur_array myarray;
BEGIN
OPEN a_cur;
LOOP
    FETCH a_cur BULK COLLECT INTO cur_array LIMIT 100;
    EXIT WHEN a_cur%NOTFOUND;
END LOOP;
CLOSE a_cur;
END;
/

DECLARE
CURSOR a_cur IS
SELECT program_id
FROM airplanes;

TYPE myarray IS TABLE OF a_cur%ROWTYPE;
cur_array myarray;
BEGIN
OPEN a_cur;
LOOP
    FETCH a_cur BULK COLLECT INTO cur_array LIMIT 500;
    EXIT WHEN a_cur%NOTFOUND;
END LOOP;
CLOSE a_cur;
END;
/

DECLARE
CURSOR a_cur IS
SELECT program_id
FROM airplanes;

TYPE myarray IS TABLE OF a_cur%ROWTYPE;
cur_array myarray;
BEGIN
OPEN a_cur;
LOOP
    FETCH a_cur BULK COLLECT INTO cur_array LIMIT 1000;
    EXIT WHEN a_cur%NOTFOUND;
END LOOP;
CLOSE a_cur;
END;
/

-- try with a LIMIT clause of 2500, 5000, and 10000. What do you see?

It used to be that you could not BULK COLLECT into an ARRAY OF RECORDS but that you could into a RECORD OF ARRAYS, as above.

This demo intentionally generated an error in versions prior to 11gR1. Familiarize yourself with the error and its message if in an earlier version. In 11gR1, and above, it is a powerful addition to the set of available tools.

CREATE OR REPLACE TYPE uw_sel_row AS OBJECT (
part_num NUMBER, part_name VARCHAR2(15));
/

CREATE OR REPLACE PROCEDURE fast_way AUTHID CURRENT_USER IS
TYPE uw_sel_tab IS TABLE OF uw_sel_row;
uw_selection uw_sel_tab;
BEGIN
SELECT uw_sel_row(part_num, part_name)
BULK COLLECT INTO uw_selection
FROM parent;

FOR i IN 1..uw_selection.COUNT LOOP
uw_selection(i).part_num := uw_selection(i).part_num * 10;
END LOOP;

FORALL i IN 1..uw_selection.COUNT
INSERT INTO child
VALUES
(uw_selection(i).part_num, uw_selection(i).part_name);
COMMIT;
END fast_way;
/

Bulk Collect with DBMS_SQL Data Types

CREATE TABLE t AS
SELECT *
FROM all_objects
WHERE 1=0;

CREATE OR REPLACE PROCEDURE nrows_at_a_time(p_array_size PLS_INTEGER) AUTHID CURRENT_USER IS
l_owner          dbms_sql.VARCHAR2_table;
l_object_name    dbms_sql.VARCHAR2_table;
l_subobject_name dbms_sql.VARCHAR2_table;
l_object_id      dbms_sql.NUMBER_table;
l_data_object_id dbms_sql.NUMBER_table;
l_object_type    dbms_sql.VARCHAR2_table;
l_created        dbms_sql.DATE_table;
l_last_ddl_time dbms_sql.DATE_table;
l_timestamp      dbms_sql.VARCHAR2_table;
l_status         dbms_sql.VARCHAR2_table;
l_temporary      dbms_sql.VARCHAR2_table;
l_generated      dbms_sql.VARCHAR2_table;
l_secondary      dbms_sql.VARCHAR2_table;

CURSOR c IS
SELECT *
FROM all_objects;
BEGIN
OPEN c;
LOOP
    FETCH c BULK COLLECT INTO
    l_owner, l_object_name, l_subobject_name, l_object_id,
    l_data_object_id, l_object_type, l_created,
    l_last_ddl_time, l_timestamp, l_status, l_temporary,
    l_generated, l_secondary
    LIMIT p_array_size;

    FORALL i in 1 .. l_owner.COUNT
      INSERT INTO t
      (owner, object_name, subobject_name, object_id,
       data_object_id, object_type, created, last_ddl_time,
       timestamp, status, temporary, generated, secondary)
      VALUES
      (l_owner(i), l_object_name(i), l_subobject_name(i),
       l_object_id(i), l_data_object_id(i),
       l_object_type(i), l_created(i), l_last_ddl_time(i),
       l_timestamp(i), l_status(i), l_temporary(i),
       l_generated(i), l_secondary(i));
    EXIT WHEN c%NOTFOUND;
END LOOP;
COMMIT;
CLOSE c;
END nrows_at_a_time;
/

Fetch First Clause

/* Limits the number of rows that a query returns, significantly reducing the SQL complexity of common "Top-N" queries. FETCH FIRST is provided primarily to simplify migration from third-party databases to Oracle Database. However, it can also improve the performance of some SELECT BULK COLLECT INTO statements*/

DECLARE
TYPE myarray IS TABLE OF airplanes.program_id%TYPE;
cur_array myarray;

n         PLS_INTEGER := 50;
BEGIN
  SELECT program_id BULK COLLECT INTO cur_array FROM airplanes FETCH FIRST n ROWS ONLY;

FOR i IN 1 .. 100 LOOP
    dbms_output.put_line(cur_array(i));
END LOOP;
END;
/

DECLARE
*
ERROR at line 1:
ORA-03113: end-of-file on communication channel
Process ID: 38552
Session ID: 17 Serial number: 2237

-- ?
DECLARE
TYPE myarray IS TABLE OF airplanes.program_id%TYPE;
cur_array myarray;
n         PLS_INTEGER := 50;
BEGIN
SELECT program_id BULK COLLECT INTO cur_array FROM airplanes WHERE ROWNUM <= n;
FOR i IN 1 .. n LOOP
    dbms_output.put_line(cur_array(i));
END LOOP;
END;
/

-- very slow and reqires hard coding of the sample size
DECLARE
TYPE myarray IS TABLE OF airplanes.program_id%TYPE;
cur_array myarray;
n         PLS_INTEGER := 50;
BEGIN
SELECT program_id BULK COLLECT INTO cur_array FROM airplanes SAMPLE(50);
FOR i IN 1 .. n LOOP
    dbms_output.put_line(cur_array(i));
END LOOP;
END;
/

FORALL

FORALL INSERT Syntax and Demos

FORALL <index_name> IN <lower_boundary> .. <upper_boundary>
<sql_statement>
SAVE EXCEPTIONS;

FORALL <index_name> IN INDICES OF <collection>
[BETWEEN <lower_boundary> AND <upper_boundary>]
<sql_statement>
SAVE EXCEPTIONS;

FORALL <index_name> IN INDICES OF <collection>
VALUES OF <index_collection>
<sql_statement>
SAVE EXCEPTIONS;

CREATE TABLE servers2 AS
SELECT *
FROM servers
WHERE 1=2;

DECLARE
CURSOR s_cur IS
SELECT *
FROM servers;

TYPE fetch_array IS TABLE OF s_cur%ROWTYPE;
s_array fetch_array;
BEGIN
OPEN s_cur;
LOOP
    FETCH s_cur BULK COLLECT INTO s_array LIMIT 1000;

    FORALL i IN 1..s_array.COUNT
    INSERT INTO servers2 VALUES s_array(i);

    EXIT WHEN s_cur%NOTFOUND;
END LOOP;
CLOSE s_cur;
COMMIT;
END;
/

FORALL UPDATE: Basic

SELECT DISTINCT srvr_id
FROM servers2
ORDER BY 1;

DECLARE
TYPE myarray IS TABLE OF servers2.srvr_id%TYPE
INDEX BY BINARY_INTEGER;

d_array myarray;
BEGIN
d_array(1) := 608;
d_array(2) := 610;
d_array(3) := 612;

FORALL i IN d_array.FIRST .. d_array.LAST
UPDATE servers2
SET srvr_id = 0
WHERE srvr_id = d_array(i);

COMMIT;
END;
/

SELECT srvr_id
FROM servers2
WHERE srvr_id = 0;

FORALL UPDATE: With SET ROW

CREATE TABLE servers3 AS
SELECT *
FROM servers
WHERE srvr_id BETWEEN 501 AND 510;

SELECT * FROM servers3;

DECLARE
CURSOR s_cur IS
SELECT *
FROM servers3;

TYPE s_rec IS TABLE OF servers3%ROWTYPE
INDEX BY PLS_INTEGER;
s_array s_rec;

-- uc = updated columns
TYPE uc_cols_rec IS RECORD(network_id servers3.network_id%TYPE);
TYPE uc_array_tt IS TABLE OF uc_cols_rec INDEX BY PLS_INTEGER;
uc_array uc_array_tt;

-- rid = row identifier
TYPE rid_t IS TABLE OF servers3.srvr_id%TYPE INDEX BY PLS_INTEGER;
rid_array rid_t;
BEGIN
OPEN s_cur;
FETCH s_cur BULK COLLECT INTO s_array;
CLOSE s_cur;

FOR i IN 1 .. s_array.COUNT LOOP
    -- load update key array
    rid_array(i) := s_array(i).srvr_id;
    -- upate values
    uc_array(i).network_id := s_array(i).network_id+1;
END LOOP;

FORALL i IN 1 .. s_array.COUNT
UPDATE (SELECT network_id FROM servers3 WHERE srvr_id = rid_array(i))
SET ROW = uc_array(i);
COMMIT;
END;
/

SELECT * FROM servers3;

FORALL DELETE

set serveroutput on

DECLARE
TYPE myarray IS TABLE OF servers2.srvr_id%TYPE
INDEX BY BINARY_INTEGER;

d_array myarray;
BEGIN
d_array(1) := 614;
d_array(2) := 615;
d_array(3) := 616;

FORALL i IN d_array.FIRST .. d_array.LAST
DELETE servers2
WHERE srvr_id = d_array(i);

COMMIT;

FOR i IN d_array.FIRST .. d_array.LAST LOOP
dbms_output.put_line('Iteration #' || i || ' deleted ' ||
SQL%BULK_ROWCOUNT(i) || ' rows.');
END LOOP;
END;
/

SELECT srvr_id
FROM servers2
WHERE srvr_id IN (614, 615, 616);

FORALL MERGE

CREATE TABLE forall_src (
rid NUMBER);

INSERT INTO forall_src VALUES (1);
COMMIT;

CREATE TABLE forall_tgt (
rid NUMBER,
ins VARCHAR2(1),
upd VARCHAR2(1));

CREATE OR REPLACE PROCEDURE forall_merge AUTHID CURRENT_USER IS
TYPE ridVal IS TABLE OF forall_tgt.rid%TYPE
INDEX BY BINARY_INTEGER;
l_data ridVal;
BEGIN
SELECT rid BULK COLLECT INTO l_data
FROM forall_src;

FORALL i IN l_data.FIRST .. l_data.LAST
MERGE INTO forall_tgt ft
USING (
    SELECT rid
    FROM forall_src fs
    WHERE fs.rid = l_data(i)) al
ON (al.rid = ft.rid)
WHEN MATCHED THEN
    UPDATE SET upd = 'U'
WHEN NOT MATCHED THEN
    INSERT (rid, ins, upd)
    VALUES (l_data(i), 'I', NULL);

COMMIT;
END forall_merge;
/

SELECT * FROM forall_src;

SELECT * FROM forall_tgt;

exec forall_merge;

SELECT * FROM forall_tgt;

exec forall_merge;

SELECT * FROM forall_tgt;

Partial Collections

Part of Collection Demo

CREATE TABLE test (
deptno NUMBER(3,0),
empname VARCHAR2(20));

INSERT INTO test VALUES (100, 'Morgan');
INSERT INTO test VALUES (200, 'Allen');
INSERT INTO test VALUES (101, 'Lofstrom');
INSERT INTO test VALUES (102, 'Havemeyer');
INSERT INTO test VALUES (202, 'Norgaard');
INSERT INTO test VALUES (201, 'Lewis');
INSERT INTO test VALUES (103, 'Scott');
INSERT INTO test VALUES (104, 'Foote');
INSERT INTO test VALUES (105, 'Townsend');
INSERT INTO test VALUES (106, 'Abedrabbo');
COMMIT;

SELECT * FROM test;

CREATE OR REPLACE PROCEDURE collection_part AUTHID CURRENT_USER IS
TYPE NumList IS VARRAY(10) OF NUMBER;
depts NumList := NumList(100,200,101,102,202,201,103,104,105,106);
BEGIN
FORALL j IN 4..7 -- use only part of varray
DELETE FROM test WHERE deptno = depts(j);
COMMIT;
END collection_part;
/

SELECT * FROM test;

Sparse Collections

A sparse collection is one from which elements have been deleted.

Sparse Collection Demo using IN INDICES OF

ALTER TABLE child
ADD CONSTRAINT uc_child_part_num
UNIQUE (part_num)
USING INDEX;

DECLARE
TYPE typ_part_name IS TABLE OF parent%ROWTYPE;
v_part typ_part_name;
BEGIN
SELECT *
BULK COLLECT INTO v_part
FROM parent;

FOR rec IN 1 .. v_part.LAST() LOOP
    IF v_part(rec).part_name != 'Rectifier' THEN
      v_part.delete(rec);
    END IF;
END LOOP;

FORALL i IN 1 .. v_part.COUNT
INSERT INTO child
VALUES
v_part(i);

COMMIT;
END;
/

DECLARE
TYPE typ_part_name IS TABLE OF parent%ROWTYPE;
v_part typ_part_name;
BEGIN
SELECT *
BULK COLLECT INTO v_part
FROM parent;

FOR rec IN 1 .. v_part.LAST LOOP
    IF v_part(rec).part_name != 'Rectifier' THEN
      v_part.delete(rec);
    END IF;
END LOOP;

FORALL idx IN INDICES OF v_part
INSERT INTO child
VALUES
v_part(idx);

COMMIT;
END;
/

SELECT COUNT(*) FROM parent;

SELECT COUNT(*) FROM child;

Using INDICES OF and VALUES OF with Non-Consecutive Index Values

CREATE TABLE valid_orders (
cust_name VARCHAR2(32),
amount NUMBER(10,2));

CREATE TABLE big_orders AS
SELECT * FROM valid_orders WHERE 1 = 0;

CREATE TABLE rejected_orders AS
SELECT * FROM valid_orders WHERE 1 = 0;

DECLARE
-- collections to hold a set of customer names and amounts
SUBTYPE cust_name IS valid_orders.cust_name%TYPE;
TYPE cust_typ IS TABLe OF cust_name;
cust_tab cust_typ;

SUBTYPE order_amount IS valid_orders.amount%TYPE;
TYPE amount_typ IS TABLE OF NUMBER;
amount_tab amount_typ;

-- collections to point into the CUST_TAB collection
TYPE index_pointer_t IS TABLE OF PLS_INTEGER;

big_order_tab index_pointer_t := index_pointer_t();

rejected_order_tab index_pointer_t := index_pointer_t();

PROCEDURE setup_data IS
BEGIN
-- set up sample order data, with some invalid and 'big' orders
cust_tab := cust_typ('Company1', 'Company2', 'Company3',
'Company4', 'Company5');

amount_tab := amount_typ(5000.01, 0, 150.25, 4000.00, NULL);
END setup_data;

BEGIN
setup_data;

dbms_output.put_line('--- Original order data ---');
FOR i IN 1..cust_tab.LAST LOOP
    dbms_output.put_line('Cust#' || i || ', '|| cust_tab(i) ||
    ': $'||amount_tab(i));
END LOOP;

-- delete invalid orders (where amount is null or 0)
FOR i IN 1..cust_tab.LAST LOOP
    IF amount_tab(i) is null or amount_tab(i) = 0 THEN
      cust_tab.delete(i);
      amount_tab.delete(i);
    END IF;
END LOOP;

dbms_output.put_line('---Data with deleted invalid orders---');

FOR i IN 1..cust_tab.LAST LOOP
    IF cust_tab.EXISTS(i) THEN
      dbms_output.put_line('Cust#' || i || ', ' || cust_tab(i) ||
      ': $'||amount_tab(i));
    END IF;
END LOOP;

-- since the subscripts of our collections are not consecutive,
-- we use use FORRALL...INDICES OF to iterate the subscripts
FORALL i IN INDICES OF cust_tab
INSERT INTO valid_orders
(cust_name, amount)
VALUES
(cust_tab(i), amount_tab(i));

-- now process the order data differently extracting
-- 2 subsets and storing each subset in a different table.
setup_data; -- reinitialize the CUST_TAB and AMOUNT_TAB collections

FOR i IN cust_tab.FIRST .. cust_tab.LAST LOOP
    IF amount_tab(i) IS NULL OR amount_tab(i) = 0 THEN
      -- add a new element to the collection
      rejected_order_tab.EXTEND;
      -- record original collection subscript
      rejected_order_tab(rejected_order_tab.LAST) := i;
    END IF;

    IF amount_tab(i) > 2000 THEN
      -- add a new element to the collection
      big_order_tab.EXTEND;
      -- record original collection subscript
      big_order_tab(big_order_tab.LAST) := i;
    END IF;
END LOOP;

-- run one DML statement on one subset of elements,
-- and another DML statement on a different subset.
FORALL i IN VALUES OF rejected_order_tab
INSERT INTO rejected_orders VALUES (cust_tab(i), amount_tab(i));

FORALL i IN VALUES OF big_order_tab
INSERT INTO big_orders VALUES (cust_tab(i), amount_tab(i));

COMMIT;
END;
/

-- verify that the correct order details were stored
SELECT cust_name "Customer", amount "Valid order amount"
FROM valid_orders;

SELECT cust_name "Customer", amount "Big order amount"
FROM big_orders;

SELECT cust_name "Customer", amount "Rejected order amount"
FROM rejected_orders;

Native Dynamic SQL

Native Dynamic SQL BULK COLLECT Statement

DECLARE
sals dbms_sql.number_table;
BEGIN
EXECUTE IMMEDIATE 'SELECT sal FROM emp'
BULK COLLECT INTO sals;
END;
/

Native Dynamic SQL FORALL Statement

CREATE TABLE tmp_target AS SELECT rownum ID, table_name, num_rows
FROM all_tables
WHERE rownum < 101;

DECLARE
TYPE NumList IS TABLE OF NUMBER;
rownos NumList;

TYPE NameList IS TABLE OF VARCHAR2(30);
tnames NameList;
BEGIN
rownos := NumList(2,4,6,8,16);

FORALL i IN 1..5
EXECUTE IMMEDIATE 'UPDATE tmp_target SET id = id * 1.1
WHERE id = :1
RETURNING table_name INTO :2'
USING rownos(i) RETURNING BULK COLLECT INTO tnames;

COMMIT;

FOR j IN 1..5 LOOP
dbms_output.put_line(tnames(j));
END LOOP;
END;
/

Combined BULK COLLECT and FORALL in a loop

-- while this demo does not require the limit clause it how to incorporate it into dynamic code

CREATE TABLE formulas (
formula_no NUMBER(1),
table_name VARCHAR2(30),
equation   CLOB);

CREATE TABLE calc_results (
formula_no NUMBER,
result_val NUMBER);

INSERT INTO formulas VALUES (1, 'SERVERS', 'AVG(latitude)');
INSERT INTO formulas VALUES (2, 'SERVERS', 'AVG(longitude)');
INSERT INTO formulas VALUES (3, 'AIRPLANES', 'COUNT(DISTINCT customer_id)');
COMMIT;

col equation format a50
SELECT * FROM formulas;

DECLARE
TYPE form_t IS TABLE OF formulas%ROWTYPE;
form_a form_t;
TYPE rset_t IS TABLE OF

TYPE rslt_t IS TABLE OF calc_results%ROWTYPE;
rslt_a rslt_t;
cSQL_Cur   INTEGER;
cur_var    SYS_REFCURSOR;
BEGIN
FOR tname_rec IN (SELECT DISTINCT table_name FROM formulas) LOOP -- 1
    BEGIN
      SELECT formula_no, table_name, equation
      BULK COLLECT INTO form_a
      FROM formulas
      WHERE table_name = tname_rec.table_name;
    END;

    DECLARE
     vSQL_Stmt   CLOB;
     cRetVal    INTEGER;
    BEGIN
      vSQL_Stmt := 'SELECT ';
      FOR j IN 1 .. form_a.COUNT LOOP    -- 2: build the statement
        vSQL_Stmt := vSQL_Stmt || form_a(j).equation || ',';
      END LOOP;                          -- 2: build the statement
      vSQL_Stmt := TRIM(TRAILING ',' FROM vSQL_Stmt);
      vSQL_Stmt := vSQL_Stmt || ' FROM ' || tname_rec.table_name;

      cSQL_Cur := dbms_sql.open_cursor;
      dbms_sql.parse(cSQL_Cur, vSQL_Stmt, dbms_sql.NATIVE);
      cRetVal := dbms_sql.execute(cSQL_Cur);
      cur_var := dbms_sql.to_refcursor(cSQL_Cur);
    END;

   DECLARE
     results    dbms_sql.number_table;
     cBatchSize CONSTANT POSITIVEN := 1000;
    BEGIN
      LOOP
        FETCH cur_var BULK COLLECT INTO results LIMIT cBatchsize;
        FOR j in 1 .. results.COUNT LOOP
          dbms_output.put_line(results(j));
      END LOOP;
        EXIT WHEN results.COUNT < cBatchSize;
     END LOOP;
    END;
END LOOP; -- 1: outer loop retrieving distinct table names
CLOSE cur_var;
END;
/

Exception Handling

Bulk Collection Exception Handling

CREATE TABLE tmp_target AS SELECT table_name, num_rows
FROM all_tables
WHERE 1=2;

ALTER TABLE tmp_target
ADD CONSTRAINT cc_num_rows
CHECK (num_rows > 0);

CREATE OR REPLACE PROCEDURE forall_errors AUTHID CURRENT_USER IS

TYPE myarray IS TABLE OF tmp_target%ROWTYPE;
l_data myarray;

CURSOR c IS
SELECT table_name, num_rows
FROM all_tables;

errors PLS_INTEGER;

array_dml EXCEPTIONS;
PRAGMA EXCEPTION_INIT(dml_errors, -24381);

BEGIN
OPEN c;
LOOP
    FETCH c BULK COLLECT INTO l_data LIMIT 100;

    -- SAVE EXCEPTIONS means don't stop if some DELETES fail
    FORALL i IN 1..l_data.COUNT SAVE EXCEPTIONS
    INSERT INTO tmp_target VALUES l_data(i);

    -- if any errors occurred during the FORALL SAVE EXCEPTIONS,
    -- a single exception is raised when the statement completes.
    EXIT WHEN c%NOTFOUND;
END LOOP;
COMMIT;
EXCEPTION
WHEN dml_errors THEN
    errors := SQL%BULK_EXCEPTIONS.COUNT;
    dbms_output.put_line('Number of DELETE statements that
    failed: ' || errors);

    FOR i IN 1 .. errors LOOP
      dbms_output.put_line('Error #' || i || ' at '|| 'iteration
      #' || SQL%BULK_EXCEPTIONS(i).ERROR_INDEX);
      dbms_output.put_line('Error message is ' ||
      SQLERRM(-SQL%BULK_EXCEPTIONS(i).ERROR_CODE));
    END LOOP;
WHEN OTHERS THEN
    RAISE;
END forall_errors;
/

SQL> exec forall_errors;

SQL> SELECT * FROM tmp_target;

Exception Handling Demo

CREATE OR REPLACE PROCEDURE array_exceptions AUTHID CURRENT_USER IS
-- cursor for processing load_errors
CURSOR le_cur IS
SELECT *
FROM load_errors
FOR UPDATE;

TYPE myarray IS TABLE OF test%ROWTYPE;
l_data myarray;

CURSOR c IS
SELECT sub_date, cust_account_id, carrier_id, ticket_id, upd_date
FROM stage
FOR UPDATE SKIP LOCKED;

errors PLS_INTEGER;

cai   test.cust_account_id%TYPE;
cid   test.carrier_id%TYPE;
ecode NUMBER;
iud   stage.upd_date%TYPE;
sd    test.sub_date%TYPE;
tid   test.ticket_id%TYPE;
upd   test.upd_date%TYPE;
BEGIN
OPEN c;
LOOP
    FETCH c BULK COLLECT INTO l_data LIMIT 50000;

    FORALL i IN 1..l_data.COUNT SAVE EXCEPTIONS
    INSERT INTO test VALUES l_data(i);

    EXIT WHEN c%NOTFOUND;
END LOOP;
COMMIT; -- exit here when no exceptions are raised
EXCEPTION
WHEN OTHERS THEN
    -- get the number of errors in the exception array
    errors := SQL%BULK_EXCEPTIONS.COUNT;

    -- insert all exceptions into the load_errors table
    FOR j IN 1 .. errors LOOP
      ecode := SQL%BULK_EXCEPTIONS(j).ERROR_CODE;
      sd := TRUNC(l_data(SQL%BULK_EXCEPTIONS(j).ERROR_INDEX).sub_date);
      cai := l_data(SQL%BULK_EXCEPTIONS(j).ERROR_INDEX).cust_account_id;
      cid := l_data(SQL%BULK_EXCEPTIONS(j).ERROR_INDEX).carrier_id;
      tid := l_data(SQL%BULK_EXCEPTIONS(j).ERROR_INDEX).ticket_id;

      INSERT INTO load_errors
      (error_code, sub_date, cust_account_id, carrier_id, ticket_id)
      VALUES
      (ecode, sd, cai, cid, tid);
    END LOOP;

    -- for each record in load_errors process those that can be handled and delete them after successful handling
    FOR le_rec IN le_cur LOOP
      IF le_rec.error_code = 1 THEN
        SELECT upd_date
        INTO iud
        FROM test
        WHERE cust_account_id = le_rec.cust_account_id
        AND carrier_id = le_rec.carrier_id
        AND ticket_id = le_rec.ticket_id;

        IF iud IS NULL THEN
          RAISE;
        ELSIF iud < le_rec.upd_date THEN
          UPDATE test
          SET upd_date = le_rec.upd_date
          WHERE sub_date = le_rec.sub_date
          AND cust_account_id = le_rec.cust_account_id
          AND carrier_id = le_rec.carrier_id
          AND ticket_id = le_rec.ticket_id;
        ELSE
          RAISE;
        END IF;
      END IF;
    END LOOP;
    COMMIT; -- exits here when any existing found
END array_exceptions;
/

Performance Demos

Performance Comparison

CREATE TABLE t1 (pnum INTEGER, pname VARCHAR2(15));
CREATE TABLE t2 AS SELECT * FROM t1;

CREATE OR REPLACE PROCEDURE perf_compare(iterations PLS_INTEGER) AUTHID CURRENT_USER IS
TYPE NumTab IS TABLE OF t1.pnum%TYPE INDEX BY PLS_INTEGER;
TYPE NameTab IS TABLE OF t1.pname%TYPE INDEX BY PLS_INTEGER;
pnums NumTab;
pnames NameTab;

a INTEGER;
b INTEGER;
c INTEGER;
BEGIN
FOR j IN 1..iterations LOOP -- load index-by tables
    pnums(j) := j;
    pnames(j) := 'Part No. ' || TO_CHAR(j);
END LOOP;

a := dbms_utility.get_time;

FOR i IN 1..iterations LOOP -- use FOR loop
    INSERT INTO t1 VALUES (pnums(i), pnames(i));
END LOOP;

b := dbms_utility.get_time;

FORALL i IN 1 .. iterations -- use FORALL statement
INSERT INTO t2 VALUES (pnums(i), pnames(i));

c := dbms_utility.get_time;

dbms_output.put_line('Execution Time (secs)');
dbms_output.put_line('---------------------');
dbms_output.put_line('FOR loop: ' || TO_CHAR((b - a)/100));
dbms_output.put_line('FORALL: ' || TO_CHAR((c - b)/100));
COMMIT;
END perf_compare;
/

set serveroutput on

exec perf_compare(500);
exec perf_compare(5000);
exec perf_compare(50000);

Bulk Collection Demo Table

CREATE TABLE parent (
part_num NUMBER,
part_name VARCHAR2(15));

CREATE TABLE child AS
SELECT *
FROM parent;

Create And Load Demo Data

DECLARE
k parent.part_name%TYPE := 'Transducer';
BEGIN
FOR i IN 1 .. 200000
LOOP
    SELECT DECODE(k, 'Transducer', 'Rectifier',
    'Rectifier', 'Capacitor',
    'Capacitor', 'Knob',
    'Knob', 'Chassis',
    'Chassis', 'Transducer')
    INTO k
    FROM dual;

    INSERT INTO parent VALUES (i, k);
END LOOP;
COMMIT;
END;
/

SELECT COUNT(*) FROM parent;
SELECT COUNT(*) FROM child;

Slow Way

CREATE OR REPLACE PROCEDURE slow_way AUTHID CURRENT_USER IS
BEGIN
FOR r IN (SELECT * FROM parent) LOOP
    -- modify record values
    r.part_num := r.part_num * 10;
    -- store results
    INSERT INTO child
    VALUES
    (r.part_num, r.part_name);
END LOOP;
COMMIT;
END slow_way;
/

set timing on

exec slow_way -- 07.71

Fast Way 1

Fetch into user defined array

CREATE OR REPLACE PROCEDURE fast_way AUTHID CURRENT_USER IS
TYPE myarray IS TABLE OF parent%ROWTYPE;
l_data myarray;

CURSOR r IS
SELECT part_num, part_name
FROM parent;

BatchSize CONSTANT POSITIVE := 1000;
BEGIN
OPEN r;
LOOP
    FETCH r BULK COLLECT INTO l_data LIMIT BatchSize;

    FOR j IN 1 .. l_data.COUNT LOOP
    l_data(j).part_num := l_data(j).part_num * 10;
END LOOP;

    FORALL i IN 1..l_data.COUNT
    INSERT INTO child VALUES l_data(i);

    EXIT WHEN l_data.COUNT < BatchSize;
END LOOP;
COMMIT;
CLOSE r;
END fast_way;
/

set timing on

exec fast_way -- 00.50

set timing off

SELECT 7.71/0.50 FROM dual;

Fast Way 2

Fetch into user defined PL/SQL table

CREATE OR REPLACE PROCEDURE fast_way AUTHID CURRENT_USER IS
TYPE PartNum IS TABLE OF parent.part_num%TYPE
INDEX BY BINARY_INTEGER;

pnum_t PartNum;

TYPE PartName IS TABLE OF parent.part_name%TYPE
INDEX BY BINARY_INTEGER;

pnam_t PartName;
BEGIN
SELECT part_num, part_name
BULK COLLECT INTO pnum_t, pnam_t
FROM parent;

FOR i IN pnum_t.FIRST .. pnum_t.LAST LOOP
pnum_t(i) := pnum_t(i) * 10;
END LOOP;

FORALL i IN pnum_t.FIRST .. pnum_t.LAST
INSERT INTO child
(part_num, part_name)
VALUES
(pnum_t(i), pnam_t(i));
COMMIT;
END fast_way;
/

set timing on

exec fast_way -- 0.62

Fast Way 3

Fetch into DBMS_SQL defined array

CREATE OR REPLACE PROCEDURE fast_way AUTHID CURRENT_USER IS
TYPE parent_rec IS RECORD (
part_num   dbms_sql.number_table,
part_name dbms_sql.varchar2_table);

p_rec parent_rec;

CURSOR c IS
SELECT part_num, part_name FROM parent;

l_done BOOLEAN;
BEGIN
OPEN c;
LOOP
    FETCH c BULK COLLECT INTO p_rec.part_num, p_rec.part_name
    LIMIT 500;
    l_done := c%NOTFOUND;

FOR i IN 1 .. p_rec.part_num.COUNT LOOP
      p_rec.part_num(i) := p_rec.part_num(i) * 10;
END LOOP;

    FORALL i IN 1 .. p_rec.part_num.COUNT
INSERT INTO child
    (part_num, part_name)
VALUES
    (p_rec.part_num(i), p_rec.part_name(i));

    EXIT WHEN (l_done);
END LOOP;
COMMIT;
CLOSE c;
END fast_way;
/

set timing on

exec fast_way -- 0.51

Fast Way 4

Affect of triggers on performance of cursor loops vs. array processing

TRUNCATE TABLE child;

set timing on

exec slow_way;
exec fast_way;

set timing off

TRUNCATE TABLE child;

CREATE OR REPLACE TRIGGER bi_child
BEFORE INSERT
ON child
FOR EACH ROW
BEGIN
NULL;
END bi_child;
/

set timing on

exec slow_way;
-- elapsed: 00:05:54.36

exec fast_way;
-- elapsed: 00:00:01.96

Fast Way 5

Insert into multiple tables

TRUNCATE TABLE child;

RENAME child TO child1;

CREATE TABLE child2 AS
SELECT * FROM child1;

CREATE OR REPLACE PROCEDURE fast_way AUTHID CURRENT_USER IS
TYPE myarray IS TABLE OF parent%ROWTYPE;
l_data myarray;

CURSOR r IS
SELECT part_num, part_name FROM parent;
BEGIN
OPEN r;
LOOP
    FETCH r BULK COLLECT INTO l_data LIMIT 1000;

    FOR j IN 1 .. l_data.COUNT LOOP
      l_data(j).part_num := l_data(j).part_num * 10;
    END LOOP;

    FORALL i IN 1..l_data.COUNT
    INSERT INTO child1 VALUES l_data(i);

    FORALL i IN 1..l_data.COUNT
    INSERT INTO child2 VALUES l_data(i);

    EXIT WHEN r%NOTFOUND;
END LOOP;
COMMIT;
CLOSE r;
END fast_way;
/

set timing on

exec fast_way

Old Way vs Fast Way with Exception Handling

CREATE TABLE tsource AS
SELECT *
FROM dba_tables;

INSERT INTO tsource
SELECT *
FROM tsource
WHERE table_name = 'DUAL';

SELECT COUNT(*)
FROM tsource;

SELECT table_name, COUNT(*)
FROM tsource
GROUP BY table_name
HAVING COUNT(*) > 1;

CREATE TABLE ttarget AS
SELECT *
FROM tsource
WHERE 1=2;

ALTER TABLE ttarget
ADD CONSTRAINT pk_ttarget
PRIMARY KEY (table_name);

CREATE OR REPLACE PROCEDURE old_way AUTHID DEFINER IS
BEGIN
FOR orec IN (SELECT * FROM tsource) LOOP
    INSERT INTO ttarget VALUES orec;
END LOOP;
COMMIT;
END old_way;
/

CREATE OR REPLACE PROCEDURE old_way AUTHID DEFINER IS
BEGIN
FOR orec IN (SELECT * FROM tsource) LOOP
    BEGIN
      INSERT INTO ttarget VALUES orec;
    EXCEPTION
      WHEN OTHERS THEN
        NULL;
    END;
END LOOP;
COMMIT;
END old_way;
/

exec old_way;
exec old_way;
exec old_way;
exec old_way;

TRUNCATE TABLE ttarget;

CREATE OR REPLACE PROCEDURE new_way AUTHID DEFINER IS
TYPE myarray IS TABLE OF tsource%ROWTYPE;
l_data myarray;

CURSOR ocur IS
SELECT *
FROM tsource;
BEGIN
OPEN ocur;
FETCH ocur BULK COLLECT INTO l_data LIMIT 1000;
FORALL i IN 1..l_data.COUNT SAVE EXCEPTIONS
INSERT INTO ttarget VALUES l_data(i);
COMMIT;
CLOSE ocur;
EXCEPTION
WHEN OTHERS THEN
    NULL;
END new_way;
/

exec new_way;
exec new_way;
exec new_way;
exec new_way;

Things To Consider

This code demonstrates a technique that is syntactically valid. But, as you will see in the second example, removes all of the benefits of using FORALL.

CREATE TABLE airplanes2 AS
SELECT * FROM airplanes
WHERE 1=2;

DECLARE
TYPE myarray IS TABLE OF airplanes2%ROWTYPE;
l_data myarray;

CURSOR r IS
SELECT *
FROM airplanes;

BatchSize CONSTANT POSITIVE := 1000;
BEGIN
OPEN r;
LOOP
    FETCH r BULK COLLECT INTO l_data LIMIT BatchSize;

    FORALL i IN 1..l_data.COUNT
    INSERT INTO airplanes2 VALUES l_data(i);

    EXIT WHEN l_data.COUNT < BatchSize;
END LOOP;
COMMIT;
CLOSE r;
END fast_way;
/

DECLARE
TYPE myarray IS TABLE OF airplanes2%ROWTYPE;
l_data myarray;

CURSOR r IS
SELECT *
FROM airplanes;

BatchSize CONSTANT POSITIVE := 1000;
BEGIN
OPEN r;
LOOP
    FETCH r BULK COLLECT INTO l_data LIMIT BatchSize;

    FORALL i IN 1..l_data.COUNT
    INSERT INTO airplanes2
    (customer_id, line_number)
    SELECT l_data(i).customer_id, l_data(i).line_number
    FROM dual;

    EXIT WHEN l_data.COUNT < BatchSize;
END LOOP;
COMMIT;
CLOSE r;
END fast_way;
/

One way to aggregate raw data. In this case summing the values of line_number

CREATE TABLE airplanes2 AS
SELECT * FROM airplanes;

INSERT INTO airplanes2
SELECT * FROM airplanes2;

INSERT INTO airplanes2
SELECT * FROM airplanes2;

set serveroutput on

DECLARE
TYPE myarray IS TABLE OF ap%ROWTYPE;
l_data myarray;
a_data myarray := myarray();

CURSOR r IS
SELECT *
FROM ap;

BatchSize CONSTANT POSITIVE := 100;
BEGIN
a_data.extend(1);
a_data(1).line_number := 0;
OPEN r;
LOOP
    FETCH r BULK COLLECT INTO l_data LIMIT BatchSize;

    FOR j IN 1 .. l_data.COUNT LOOP
      a_data(1).line_number := a_data(1).line_number + l_data(j).line_number;
    END LOOP;
    EXIT WHEN l_data.COUNT < BatchSize;
END LOOP;
dbms_output.put_line(a_data(1).line_number);
END;
/