并行管道函数

sys@JINGYONG> create table t1  2  as  3  select object_id id,object_name text  4  from all_objects;表已创建。sys@JINGYONG> begin  2  dbms_stats.set_table_stats  3  (user,'T1',numrows=>100000000,numblks=>100000);  4  end;  5  /PL/SQL 过程已成功完成。sys@JINGYONG> create table t2  2  as  3  select t1.*,0 session_id  4  from t1  5  where 1=0;表已创建。

这里使用DBMS_STATS来骗过优化器，让它以为输入表中有10,000,000行，而且占用了100,000个数据库块。在此模拟 一个大表。第二个表T2是第一个表的一个副本，只是在结构中增加了一个SESSION_ID列。可以通过它具体看到是否发生了并行化。接下来，需要建立管道函数返回的对象类型。在这个例子中，对象类型类似于T2：

sys@JINGYONG> create or replace type t2_type  2  as object  3  (  4  id number,  5  text varchar2(30),  6  session_id number  7  );  8  /类型已创建。sys@JINGYONG> create or replace type t2_tab_type as table of t2_type;  2  /类型已创建。

现在这个过程是一个生成行的函数。它接收数据作为输入，并在一个引用游标（ref cursor）中处理。这个函数返回一个 T2_TAB_TYPE，这就是我们刚才创建的对象类型。这是一个PARALLEL_ENABLED（启用子并行）的管道函数。在此使用了分区 （partition）子句，这就告诉Oracle：以任何最合适的方式划分或分解数据。我们不需要对数据的顺序做任何假设。

在此，我们只想划分数据。数据如何划分对于我们的处理并不重要，所以定义如下：

sys@JINGYONG> create or replace function parallel_pipelined(l_cursor in sys_refcursor)  2  return t2_tab_type  3  pipelined  4  parallel_enable(partition l_cursor by any)  5  is  6   l_session_id number;  7   TYPE type_t1_data IS TABLE OF t1%ROWTYPE INDEX BY PLS_INTEGER;  8   l_t1  type_t1_data;  9 10  begin 11  select sid into l_session_id 12  from v$mystat 13  where rownum=1; 14  loop 15    fetch l_cursor bulk collect into l_t1;--用bulk collect来一次性获取数据 16    exit when l_t1.count=0; 17    for i in 1 .. l_t1.count loop 18          pipe row(t2_type(l_t1(i).id,l_t1(i).text,l_session_id)); 19    end loop; 20    null; 21  end loop; 22  close l_cursor; 23  return; 24  end; 25  /Function created

或者用下面的过程来一行一行来获取

create or replacefunction parallel_pipelined( l_cursor in sys_refcursor )return t2_tab_typepipelinedparallel_enable ( partition l_cursor by any )is l_session_id number; l_rec t1%rowtype;begin select sid into l_session_id from v$mystat where rownum =1; loop fetch l_cursor into l_rec; exit when l_cursor%notfound; pipe row(t2_type(l_rec.id,l_rec.text,l_session_id)); end loop; close l_cursor; return;end;

这样就创建了函数。我们准备并行地处理数据，让Oracle根据可用的资源来确定最合适的并行度：

SQL> insert /*+ append */  2  into t2(id,text,session_id)  3   select *  4   from table(parallel_pipelined  5   (CURSOR(select /*+ parallel(t1) */ *  6   from t1 )  7  ))  8  ;50333 rows insertedSQL> commit;Commit complete

为了查看这里发生了什么，可以查询新插入的数据，并按SESSION_ID分组，先来看使用了多少个并行执行服务器，再看每个并行 执行服务器处理了多少行：

SQL> select session_id,count(*) from t2 group by session_id;SESSION_ID   COUNT(*)---------- ----------       136      31006       145      19327

显然，对于这个并行操作的SELECT部分，我们使用了2个并行执行服务器,可以看到，Oracle对我们的过程进行了并行化