Do-It-Yourself Parallelism

Say we have that same process as in the preceding section: the serial, simple procedure. We cannot afford a rather extensive rewrite of the implementation, but we would like to execute it in parallel. What can we do? 

Starting with Oracle Database 11g Release 2 and above – we have a new way to implement parallelism via the DBMS_PARALLLEL_EXECUTE builtin package.  Using it, you can execute a SQL or PL/SQL statement in parallel by taking the data to be processed and breaking it up into multiple, smaller streams. You can implement the same approach yourself, manually, and I show how in the section on “Old School Do-It-Yourself Parallelism” that follows. The beauty of the new package though, is that it eliminates much of the tedious work that you otherwise need to perform.

Let’s start with the premise that we have a SERIAL routine, that we’d like to execute in parallel against some large table.  We’d like to do it with as little work as possible – modify as little code as possible and be responsible for generating very little new code.  Enter DBMS_PARALLEL_EXECUTE.  We will not cover every possible use of this package (it is fully documented in the Oracle® Database PL/SQL Packages and Types Reference 11g Release 2 (11.2) guide) – but rather use just enough of it to implement the process I’ve just described.

Assuming we start with the empty table T2 – we’ll modify our serial process now to look like this – additions to the original, simple serial process are in bold:

big_table%ORA11GR2> create or replace
  2  procedure serial( p_lo_rid in rowid, p_hi_rid in rowid )
  3  is
  4  begin
  5      for x in ( select object_id id, object_name text
  6                   from big_table
  7                  where rowid between p_lo_rid
  8                                  and p_hi_rid )
  9      loop
 10          -- complex process here
 11          insert into t2 (id, text, session_id )
 12          values ( x.id, x.text, sys_context( 'userenv', 'sessionid' ) );
 13      end loop;
 14  end;
 15  /

Procedure created.

That was it – just add the ROWID inputs and the predicate.  The modified code has not changed much at all.  I am using SYS_CONTEXT to get the SESSIONID so we can monitor how much work was done by each ‘thread’, each parallel session.

Now, to start the process – we first need to break up the table into small pieces.  We can do this by some numeric range – useful for tables that use a SEQUENCE to populate their primary key, by any arbitrary SQL you want to code, or by ROWID ranges.  We’ll use the ROWID range, I find that to simply be the most efficient – it creates non-overlapping ranges of the table (contention free) and doesn’t require querying the table to decide the ranges, it just uses the data dictionary.  So, we’ll make the following API calls:

big_table%ORA11GR2> begin
  2      dbms_parallel_execute.create_task(‘PROCESS BIG TABLE’);
  3      dbms_parallel_execute.create_chunks_by_rowid
  4      ( task_name   => ‘PROCESS BIG TABLE’,
  5        table_owner => user,
  6        table_name  => 'BIG_TABLE',
  7        by_row      => false,
  8        chunk_size  => 10000 );
  9  end;
 10  /
PL/SQL procedure successfully completed.

We started by creating a named task – ‘PROCESS BIG TABLE’ in this case.  That is just a unique name we’ll use to refer to our big process.  Second we invoked the CREATE_CHUNKS_BY_ROWID procedure.  That procedure does exactly what its name implies, it ‘chunks up’ a table by ROWID ranges in a manner similar to what we just did above.  We told the procedure to read the information about the currently logged in user’s table named BIG_TABLE and to break it up into chunks of no more than about 10,000 blocks (CHUNK_SIZE).  The parameter BY_ROW was set to false which implies in this case that the CHUNK_SIZE is not a count of rows to create ROWID ranges by, but rather a count of blocks to create them.

We can see the number of chunks and information about each chunk immediately after this block of code executes by querying DBA_PARALLEL_EXECUTE_CHUNKS – a new view:

big_table%ORA11GR2> select *
  2    from (
  3  select chunk_id, status, start_rowid, end_rowid
  4    from dba_parallel_execute_chunks
  5   where task_name = ‘PROCESS BIG TABLE’
  6   order by chunk_id
  7         )
  8   where rownum <= 5
  9  /

  CHUNK_ID STATUS               START_ROWID        END_ROWID
---------- -------------------- ------------------ ------------------
       590 UNASSIGNED           AAAXZ0AAEAAAAKAAAA AAAXZ0AAEAAAAZ/CcP
       591 UNASSIGNED           AAAXZ0AAEAAAAsAAAA AAAXZ0AAEAAAAt/CcP
       592 UNASSIGNED           AAAXZ0AAEAAAAuAAAA AAAXZ0AAEAAAAv/CcP
       593 UNASSIGNED           AAAXZ0AAEAAAAwAAAA AAAXZ0AAEAAAAx/CcP
       594 UNASSIGNED           AAAXZ0AAEAAAAyAAAA AAAXZ0AAEAAAAz/CcP

The query in this example shows the first five rows in the view – in my case there were 209 total rows in there for the table in question, each representing a non-overlapping chunk of the table to process.  This does not mean we’ll be processing the table in “parallel 209” – but just that we have 209 chunks in total to process.  We are now ready to run our task, that is accomplished via this API call:

big_table%ORA11GR2> begin
  2      dbms_parallel_execute.run_task
  3      ( task_name      => ‘PROCESS BIG TABLE’,
  4        sql_stmt       => 'begin serial( :start_id, :end_id ); end;',
  5        language_flag  => DBMS_SQL.NATIVE,
  6        parallel_level => 4 );
  7  end;
  8  /
PL/SQL procedure successfully completed.

Here we asked to run our task ‘PROCESS BIG TABLE’ – which points to our chunks.  The SQL statement we want to execute is 'begin serial( :start_id, :end_id ); end;' – a simple call to our stored procedure with the ROWID range to process.  The PARALLEL_LEVEL I decided to use was four – meaning, we’ll have four parallel threads/processes executing this.  Even though there were 209 chunks – we’ll only do four at a time.  Internally, this package uses the DBMS_SCHEDULER package to run these threads in parallel.  

Once our task starts running – it will create four jobs, and each job is told to process the chunks identified by the key value ‘PROCESS BIG TABLE’ and run the stored procedure SERIAL against each chunk.  So, these four sessions start and each reads a chunk from the DBA_PARALLEL_EXECUTE_CHUNKS view, processes it, and updates the STATUS column.  If the chunk is successful – the row will be marked as PROCESSED, if it fails for any reason – if a given chunk cannot be processed, it will be marked as PROCESSED_WITH_ERROR and other columns will contain the detailed error message indicating the cause of the error.  In either case – the session will then retrieve another chunk and process it and so on.  So, eventually these four jobs will have processed all of the chunks and the task will complete.

If any of the chunks failed, you can correct the underlying cause of the error and resume the task.  That will cause it to reprocess the failed chunks.  When all complete successfully:

big_table%ORA11GR2> select *
  2    from (
  3  select chunk_id, status, start_rowid, end_rowid
  4    from dba_parallel_execute_chunks
  5   where task_name = 'process big table'
  6   order by chunk_id
  7         )
  8   where rownum <= 5
  9  /

  CHUNK_ID STATUS               START_ROWID        END_ROWID
---------- -------------------- ------------------ ------------------
       590 PROCESSED            AAAXZ0AAEAAAAKAAAA AAAXZ0AAEAAAAZ/CcP
       591 PROCESSED            AAAXZ0AAEAAAAsAAAA AAAXZ0AAEAAAAt/CcP
       592 PROCESSED            AAAXZ0AAEAAAAuAAAA AAAXZ0AAEAAAAv/CcP
       593 PROCESSED            AAAXZ0AAEAAAAwAAAA AAAXZ0AAEAAAAx/CcP
       594 PROCESSED            AAAXZ0AAEAAAAyAAAA AAAXZ0AAEAAAAz/CcP

You are done. You can either retain the task for “history” or remove it. The following example shows how to remove the task:

big_table%ORA11GR2> begin
  2        dbms_parallel_execute.drop_task('process big table' );
  3  end;
  4  /
PL/SQL procedure successfully completed.

If we review our own application table – we can see that in fact the job was done using parallel 4 – and each of the four processed about the same number of rows:

big_table%ORA11GR2> select session_id, count(*)
  2    from t2
  3   group by session_id
  4   order by session_id;

SESSION_ID   COUNT(*)
---------- ----------
  21855337    2412538
  21855338    2478642
  21855339    2639919
  21855340    2468901

If you aren’t yet up to Oracle Database 11g Release 2, then you can implement the same sort of parallelism using the more labor-intensive approach in the next section. However, the new package provides a rich API (which we’ve just touched on here) that has much more functionality than the manual implementation does.




Old School Do-It-Yourself Parallelism

Prior to Oracle Database 11g Release 2, you can implement a parallelization approach similar to that in the previous section. You won’t have a rich API to support you, and you’ll need to do more of the tedious work yourself, but it can be done. My approach many times has been to use rowid ranges to break the table up into some number of ranges that don’t overlap (yet completely cover the table).
Note: If you are running Oracle Database 11g Release 2 or higher, refer to the previous section. There you’ll find an example showing the use of the DBMS_PARALLLEL_EXECUTE package. If it’s available to you, you really should be using that package rather than the manual approach described here.

This manually-intensive approach is very similar to how Oracle performs a parallel query conceptually. If you think of a full table scan, Oracle processes that by coming up with some method to break the table into many “small” tables, each of which is processed by a parallel execution server. We are going to do the same thing using rowid ranges. In early releases, Oracle’s parallel implementation actually used rowid ranges itself.

Again, we’ll use a BIG_TABLE of 1,000,000 rows, as the technique I’m describing works best on big tables with lots of extents, and the method I use for creating rowid ranges depends on extent boundaries. The more extents used, the better the data distribution. So, after creating the BIG_TABLE with 1,000,000 rows, we’ll create T2 like this:


big_table@ORA11GR2> create table t2
  2  as
  3  select object_id id, object_name text, 0 session_id
  4    from big_table
  5   where 1=0;
Table created.

We are going to use the job queues built into the database to parallel process our procedure. We will schedule some number of jobs. Each job is our procedure slightly modified to just process the rows in a given rowid range.
Note In Oracle 10g and above, you could use the scheduler as well for something so simple.  In order to make the example 9i compatible, we’ll use the job queues here.

To efficiently support the job queues, we’ll use a parameter table to pass inputs to our jobs:

big_table@ORA11GR2> create table job_parms
  2  ( job        number primary key,
  3    lo_rid  rowid,
  4    hi_rid  rowid
  5  )
  6  /
Table created.


This will allow us to just pass the job ID into our SERIAL procedure, so it can query this table to get the rowid range it is to process. Now, for our procedure. The code in bold is the new code we’ll be adding:


big_table@ORA11GR2> create or replace
  2  procedure serial( p_job in number )
  3  is
  4      l_rec        job_parms%rowtype;
  5  begin
  6      select * into l_rec
  7        from job_parms
  8       where job = p_job;
  9
 10      for x in ( select object_id id, object_name text
 11                   from big_table
 12                  where rowid between l_rec.lo_rid
 13                                  and l_rec.hi_rid )
 14      loop
 15          -- complex process here
 16          insert into t2 (id, text, session_id )
 17          values ( x.id, x.text, p_job );
 18      end loop;
 19
 20      delete from job_parms where job = p_job;
 21      commit;
 22  end;
 23  /
Procedure created.


As you can see, it is not a significant change. Most of the added code was simply to get our inputs and the rowid range to process. The only change to our logic was the addition of the predicate on lines 12 and 13.

Now let’s schedule our job. We’ll use a rather complex query using analytics to divide the table. The innermost query on lines 19 through 26 breaks the data into eight groups in this case. The first sum on line 22 is computing a running total of the sum of blocks; the second sum on line 23 is the total number of blocks. If we integer divide the running total by the desired “chunk size” (the total size divided by 8 in this case), we can create groups of files/blocks that cover about the same amount of data. The query on lines 8 through 28 finds the high and low file numbers and block numbers by GRP and returns the distinct entries. It builds the inputs we can then send to DBMS_ROWID to create the rowids Oracle wants. We take that output and, using DBMS_JOB, submit a job to process the rowid range: 

big_table@ORA11GR2> declare
  2          l_job number;
  3  begin
  4  for x in (
  5  select dbms_rowid.rowid_create
            ( 1, data_object_id, lo_fno, lo_block, 0 ) min_rid,
  6         dbms_rowid.rowid_create
            ( 1, data_object_id, hi_fno, hi_block, 10000 ) max_rid
  7    from (
  8  select distinct grp,
  9         first_value(relative_fno) 
              over (partition by grp order by relative_fno, block_id
 10           rows between unbounded preceding and unbounded following) lo_fno,
 11         first_value(block_id    ) 
              over (partition by grp order by relative_fno, block_id
 12           rows between unbounded preceding and unbounded following) lo_block,
 13         last_value(relative_fno) 
              over (partition by grp order by relative_fno, block_id
 14           rows between unbounded preceding and unbounded following) hi_fno,
 15         last_value(block_id+blocks-1) 
              over (partition by grp order by relative_fno, block_id
 16           rows between unbounded preceding and unbounded following) hi_block,
 17         sum(blocks) over (partition by grp) sum_blocks
 18    from (
 19  select relative_fno,
 20         block_id,
 21         blocks,
 22         trunc( (sum(blocks) over (order by relative_fno, block_id)-0.01) /
 23                (sum(blocks) over ()/8) ) grp
 24    from dba_extents
 25   where segment_name = upper('BIG_TABLE')
 26     and owner = user order by block_id
 27         )
 28         ),
 29         (select data_object_id 
               from user_objects where object_name = upper('BIG_TABLE') )
 30  )
 31  loop
 32          dbms_job.submit( l_job, 'serial(JOB);' );
 33          insert into job_parms(job, lo_rid, hi_rid)
 34          values ( l_job, x.min_rid, x.max_rid );
 35  end loop;
 36  end;
 37  /
PL/SQL procedure successfully completed. 

That PL/SQL block would have scheduled up to eight jobs for us (fewer if the table could not be broken into eight pieces due to insufficient extents or size). We can see how many jobs were scheduled and what their inputs were as follows:

big_table@ORA11GR2> select * from job_parms;

       JOB LO_RID             HI_RID
---------- ------------------ ------------------
       172 AAAT7tAAEAAAAkpAAA AAAT7tAAEAAABQICcQ
       173 AAAT7tAAEAAABQJAAA AAAT7tAAEAAABwICcQ
       174 AAAT7tAAEAAABwJAAA AAAT7tAAEAAACUICcQ
       175 AAAT7tAAEAAACUJAAA AAAT7tAAEAAAC0ICcQ
       176 AAAT7tAAEAAAC0JAAA AAAT7tAAEAAADMICcQ
       177 AAAT7tAAEAAADaJAAA AAAT7tAAEAAAD6ICcQ
       178 AAAT7tAAEAAAD6JAAA AAAT7tAAEAAAEaICcQ
       179 AAAT7tAAEAAAEaJAAA AAAT7tAAEAAAF4ICcQ
8 rows selected.

big_table@ORA11GR2> commit;
Commit complete.

That commit released our jobs for processing. We have JOB_QUEUE_PROCESSES set to 1000 in the parameter file, so all eight started running and shortly finished. After they all completed, the results are as follows:

big_table%ORA11GR2> select session_id, count(*)
  2    from t2
  3   group by session_id;

SESSION_ID   COUNT(*)
---------- ----------
         6    1318540
        11    1298502
         5    1259102
         4    1111874
         8    1291791
         7    1152626
        10    1284181
         9    1283384

8 rows selected.

Suppose, however, that you do not want to use the rowid processing—perhaps the query is not as simple as SELECT * FROM T and involves joins and other constructs that make using the rowid impractical. You can use the primary key of some table instead. For example, say you want to break that same BIG_TABLE into ten pieces to be processed concurrently by primary key. You can do that easily using the NTILE built-in analytic function. The process is rather straightforward:


big_table@ORA11GR2> select nt, min(id), max(id), count(*)
  2    from (
  3  select id, ntile(10) over (order by id) nt
  4    from big_table
  5         )
  6   group by nt;

        NT    MIN(ID)    MAX(ID)   COUNT(*)
---------- ---------- ---------- ----------
         1          1     100000     100000
         2     100001     200000     100000
         3     200001     300000     100000
         4     300001     400000     100000
         5     400001     500000     100000
         6     500001     600000     100000
         7     600001     700000     100000
         8     700001     800000     100000
         9     800001     900000     100000
        10     900001    1000000     100000

10 rows selected.

Now you have ten nonoverlapping primary key ranges, all of nice equal size, that you can use to implement the same DBMS_JOB technique as shown earlier to parallelize your process.