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 function 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 or insert as select. Here’s some sample code (tested on 11.1.0.6) to demonstrate the principle:

-- 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:

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

Once I’ve created these object 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 objects 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 target table as it selects it. The reason becomes clear as 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 allowed us to push the 60,000 rows, join and aggregate remotely and 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 stuff like that 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 at the same client at the time and the team that  needed to solve this particular performance problem sent members to both of us for help – and got the same suggested solution from both of us.