11g: select with union all from two different table function not parallel

From: Frank Bergemann <FBergemann_at_web.de>
Date: Mon, 20 Feb 2012 04:50:13 -0800 (PST)
Message-ID: <a65fdde8-c494-4626-a47b-2e5856b220eb_at_p7g2000yqk.googlegroups.com>

Hi,

i posted it here: https://forums.oracle.com/forums/thread.jspa?messageID=10152723 But after submitting some demo-scripts i didn't get any feedback(?!).

The headline was:

---

insert into ...
select * from (

    select * from TableFunc1(...)
    union all
    select * from TableFunc2(...)
);

... does not start any TableFunc2 instance before all TableFunc1 instances have finished.

I thought 'union all' will not sequentialize the two queries(?!) Or do i miss something else to make it work in parallel?

---

A demo script here:

• snip ---------------------------------- drop table Test_Table;
  /
  create table Test_Table( sid number, name varchar2(10), t1 timestamp with time zone, t2 timestamp with time zone, value number);
  /
  

alter table Test_Table parallel (degree 5);
/

drop type Test_List;
/

create or replace type Test_Object as object (   sid number,
  name varchar2(10),
  t1 timestamp with time zone,
  t2 timestamp with time zone,
  value number);
/

create or replace type Test_List is table of Test_Object;
/

create or replace package Test_Pkg
as

  type Test_Rec is record(
    sid number,
    name varchar2(10),
    t1 timestamp with time zone,
    t2 timestamp with time zone,
    value number);

  Type Test_tab is table of test_Rec;

  type Test_Cur is ref cursor return Test_Rec;

  function TF1(
    cur in Test_Cur)
  return Test_Tab pipelined
  parallel_enable (partition cur by any);

  function TF2(
   cur in Test_Cur)
  return Test_Tab pipelined
  parallel_enable (partition cur by any);

  procedure do;

end;
/

create or replace package body Test_Pkg
as

  function TF1(
    cur in Test_Cur)
  return Test_Tab pipelined
  parallel_enable (partition cur by any)   is
    testTab Test_Tab;
    rec test_rec;
  begin
    select userenv('SID') into rec.sid from dual;     rec.name := 'TF1';
    rec.t1 := current_timestamp;
    dbms_lock.sleep(3);
    loop

      fetch cur bulk collect into testTab;
      exit when testTab.count = 0;
      for i in testTab.first..testTab.last
      loop
        rec.value := testTab(i).value;
        rec.t2 := current_timestamp;
        pipe row (rec);
      end loop;

    end loop;
  end;

  function TF2(
    cur in Test_Cur)
  return Test_Tab pipelined
  parallel_enable (partition cur by any)   is
    testTab Test_Tab;
    rec test_rec;
  begin
    select userenv('SID') into rec.sid from dual;     rec.name := 'TF2';
    rec.t1 := current_timestamp;
    dbms_lock.sleep(5);
    loop

      fetch cur bulk collect into testTab;
      exit when testTab.count = 0;
      for i in testTab.first..testTab.last
      loop
        rec.value := testTab(i).value;
        rec.t2 := current_timestamp;
        pipe row (rec);
      end loop;

    end loop;
  end;

  procedure Do
  is
    sid number;

    tf1_t1_min timestamp with time zone;
    tf1_t2_max timestamp with time zone;
    tf2_t1_min timestamp with time zone;
    tf2_t2_max timestamp with time zone;

    type sid_hash is table of number index by binary_integer;     tf1_sids sid_hash;
    tf2_sids sid_hash;
    strBuff varchar2(1024);

    procedure logSids(sids sid_Hash)
    is

      i binary_integer;
      strBuff varchar2(4096);
    begin
      i := sids.first;
      while i <= sids.last
      loop
        strBuff := strBuff || ', ' || i;
        i := sids.next(i);
      end loop;
      dbms_output.put_line('  #' ||sids.count || ' sids : ' ||

strBuff);

    end;

  begin
    select userenv('SID') into sid from dual;     dbms_output.put_line('main thread sid is #' || sid);     for i in 1..10000
    loop
      insert into Test_Table values (null, null, null, null, i);     end loop;
    commit;
    for rec in (
      select * from table(TF1(cursor(select /*+ first_rows dynamic_sampling(t1) */ * from Test_Table))) t1

      union all
      select * from table(TF2(cursor(select /*+ first_rows

dynamic_sampling(t2) */* from Test_Table))) t2

    )
    loop

      if rec.name='TF1'
      then
        tf1_sids(rec.sid) := 1;
        if tf1_t1_min is null or rec.t1 < tf1_t1_min then
tf1_t1_min := rec.t1; end if;
        if tf1_t2_max is null or rec.t2 > tf1_t2_max then
tf1_t2_max := rec.t2; end if;
      else -- 'TF2'
        tf2_sids(rec.sid) := 1;
        if tf2_t1_min is null or rec.t1 < tf2_t1_min then
tf2_t1_min := rec.t1; end if;
        if tf2_t2_max is null or rec.t2 > tf2_t2_max then
tf2_t2_max := rec.t2; end if;
      end if;
    end loop;
      dbms_output.put_line('TF1');
      dbms_output.put_line(' t1(min) = ' || to_char(tf1_t1_min,
'HH:MI::SS') || ' , t2(max) = ' || to_char(tf1_t2_max, 'HH:MI::SS'));
      logSids(tf1_sids);
      dbms_output.put_line('TF2');
      dbms_output.put_line(' t1(min) = ' || to_char(tf2_t1_min,
'HH:MI::SS') || ' , t2(max) = ' || to_char(tf2_t2_max, 'HH:MI::SS'));
      logSids(tf2_sids);

  end;
end;
/

set serveroutput on

declare
begin
  test_pkg.do;
end;

------------- snap ----------------------------------

Output is:

------------- snip ----------------------------------

main thread sid is #8
TF1
 t1(min) = 04:44::09 , t2(max) = 04:44::12   #5 sids : , 16, 51, 76, 117, 172
TF2
 t1(min) = 04:44::12 , t2(max) = 04:44::17   #5 sids : , 16, 51, 76, 117, 172

------------- snap ----------------------------------

Can somebody at least confirm this behavior? Preferable tell me, if there is a way to make it work :-)

• many thanks!

best regards,
Frank Received on Mon Feb 20 2012 - 06:50:13 CST