Distributed Pipelines

In an article that I wrote about the  driving_site() hint a few months ago I pointed out that the hint was not supposed to work with “create table as select” (CTAS) and “insert as select”. One of the people commenting on the note mentioned pipelined functions as a workaround to this limitation – and I’ve finally got around to writing a note about the method.

The idea is simple. If you can write a distributed select statement that takes advantage of the /*+ driving_site(alias) */ hint to work efficiently you can wrap the statement in a PL/SQL cursor loop and stick that loop into a pipelined function to maximise the efficiency of create/insert as select. Here’s some sample code (tested on 11.1.0.6) to demonstrate the principle:

rem
rem     Script:         dist_pipe.sql
rem     Author:         Jonathan Lewis 
rem     Purpose:        
rem     Dated:          Sept 2010
rem
rem     Last tested 
rem             11.1.0.6
rem

-- create public database link d11g@loopback using 'd11g';

create table t1 as
select
        rownum                  id,
        lpad(rownum,15,'x')     val
from
        all_objects
where
        rownum <= 1000
;

create table t2 as
select
        rownum                  id,
        lpad(rownum,15,'y')     val
from
        all_objects
where
        rownum <= 1000
;

create table t3(
        t1_val  varchar2(15),
        t2_val  varchar2(15)
)
;

--      Collect stats on the tables at this point.

create or replace type myScalarType as object (
        t1_val  varchar2(15),
        t2_val  varchar2(15)
)
/

create or replace type myArrayType as table of myScalarType
/

create or replace function pipe_fun
return myArrayType
pipelined
as
begin
        for r1 in (
                select
                        /*+ driving_site(t2) */
                        t1.val  t1_val,
                        t2.val  t2_val
                from
                        t1                      t1,
                        t2@d11g@loopback        t2
                where
                        t2.id = t1.id
        ) loop
                pipe row (myScalarType(r1.t1_val, r1.t2_val));
        end loop;
        return;
end;
/

I’ve set up my tnsnames.ora and created a database link that I can use to query the “remote” database. If you’re familiar with database links you’ll probably guess that I’ve actually created a loopback link, so the same database is acting as both the local and remote databases.

After creating and populating a couple of source tables and creating an empty target table for the results of a query, I’ve created the infrastructure to support a pipelined function matching the target table.

I need two things:

• a scalar object type that looks like a row from the table
• a table type which is a table of that scalar object type.

Once I’ve created these two types I can create a function that can return the table type, which it does by “piping” items of the scalar type. Note, particularly, that a pipelined function doesn’t “return” anything, instead it pipes rows  of the supporting scalar type. (It would be legal, by the way, to use the “pipe row” more than once in the definition of the function – which you might do if, for example, you wanted to use a pipelined function for normalising some data that you were reading from a flat file that had been declared as an external table.)

At this point you might be wondering why I’ve created the pipelined function at all, and why I haven’t simply created a procedure that inserts the data into the target table as it selects it. The reason becomes clear when I demonstrate how I can use the pipelined function:

alter table t3 nologging;

insert /*+ append */ into t3
select
        *
from
        table(pipe_fun)
;

commit;

alter table t3 logging;

The function can behave like a table and I can get array processing speeds (plus the benefits of direct path and nologging actions) by selecting from that “table”. So not only have I bypassed the limitation of the /*+ driving_site(alias) */ hint, I’ve done it in a way that gives me a number of options for making the process as efficient as possible.

I’ve been a little lazy here, of course. My function is using a single cursor “row by row” select before I call it to supply data for the  insert so, arguably, I ought to enhance the function to use the pl/sql “bulk collect” option, perhaps with a “limit” clause. I haven’t done so because my client at the time was using 11g and (from 10g onwards) the pl/sql compiler will automatically turn my cursor loop code into the “bulk collect” equivalent with a limit of 100 rows. As the first draft of this code reduced the client’s run time from an estimated 8 days to 288 seconds (and that was without using the nologging/append options on the target table) we didn’t feel there was any need to squeeze out the extra bit of performance.

If you’re wondering why the performance benefit could be so large – it’s the sort of thing that happens when you’ve got 60,000 rows in the local table and 250,000,000 in the remote and need to join and aggregate. The last thing you want to do a nested loop with 60,000 indexed probes across the network, and the second to last thing is to pull the entire 250,000,000 rows across the network to join and aggregate locally. The pipelined function gave us a method of pushing the 60,000 rows to the remote instance to join and aggregate there, then pull a small (60,000 row) result set back.

If you want to see the effects on execution plans – here are the two plans I pulled from memory. The first is the SQL run by the function before I added the /*+ driving_site(t2) */ hint to the query; the second is the SQL that arrived at the remote database when the hint was in place.

SQL_ID  3v0qf8uajpsc8, child number 0
-------------------------------------
SELECT T1.VAL T1_VAL, T2.VAL T2_VAL FROM T1, T2@D11G@LOOPBACK WHERE
T2.ID = T1.ID

Plan hash value: 1770389500

-------------------------------------------------------------------------------------------
| Id  | Operation          | Name | Rows  | Bytes | Cost (%CPU)| Time     | Inst   |IN-OUT|
-------------------------------------------------------------------------------------------
|   0 | SELECT STATEMENT   |      |       |       |     6 (100)|          |        |      |
|*  1 |  HASH JOIN         |      |  1000 | 38000 |     6  (17)| 00:00:01 |        |      |
|   2 |   TABLE ACCESS FULL| T1   |  1000 | 19000 |     3   (0)| 00:00:01 |        |      |
|   3 |   REMOTE           | T2   |  1000 | 19000 |     2   (0)| 00:00:01 | D11G@~ | R->S |
-------------------------------------------------------------------------------------------

Predicate Information (identified by operation id):
---------------------------------------------------
   1 - access("T2"."ID"="T1"."ID")

Remote SQL Information (identified by operation id):
----------------------------------------------------
   3 - SELECT "ID","VAL" FROM "T2" "T2" (accessing 'D11G@LOOPBACK' )

SQL_ID  csgj6635jzj0u, child number 0
-------------------------------------
SELECT "A2"."VAL","A1"."VAL" FROM "T1"@! "A2","T2" "A1" WHERE
"A1"."ID"="A2"."ID"

Plan hash value: 1827050725

-------------------------------------------------------------------------------------------
| Id  | Operation          | Name | Rows  | Bytes | Cost (%CPU)| Time     | Inst   |IN-OUT|
-------------------------------------------------------------------------------------------
|   0 | SELECT STATEMENT   |      |       |       |     6 (100)|          |        |      |
|*  1 |  HASH JOIN         |      |  1000 | 38000 |     6  (17)| 00:00:01 |        |      |
|   2 |   REMOTE           | T1   |  1000 | 19000 |     2   (0)| 00:00:01 |      ! | R->S |
|   3 |   TABLE ACCESS FULL| T2   |  1000 | 19000 |     3   (0)| 00:00:01 |        |      |
-------------------------------------------------------------------------------------------

Predicate Information (identified by operation id):
---------------------------------------------------
   1 - access("A1"."ID"="A2"."ID")

Remote SQL Information (identified by operation id):
----------------------------------------------------
   2 - SELECT "ID","VAL" FROM "T1" "A2" (accessing '!' )

You’ll notice that the distributed SQL is in capitals with lots of double-quote marks and table aliases like A1, A2 – If you ever see SQL matching that pattern there’s a reasonable chance that you’re seeing incoming distributed SQL or (on older versions of Oracle) the decomposed text of parallel execution.

Footnote: I was quite amused by the fact that Peter Scott was working in another department at the same client at the time, and the team that  needed to solve this particular performance problem had sent members to both of us for help – and got the same suggested solution from both of us.