管道函数

  TYPE datetime_range_list IS TABLE OF datetime_range_record;
  --当前工作时间下，产线的出勤信息（包括当前班组、出勤起止时间等）
  TYPE record_attendance_info IS RECORD(
    workgroup_arrange_id      VARCHAR2(32), --班组排班ID
    resource_calendar_id      VARCHAR2(32), --资源日历ID
    workgroup_id              VARCHAR2(32), --班组ID
    attendance_id             VARCHAR2(32), --出勤模式ID
    shift_code                VARCHAR2(80), --班次编码
    shift_date                DATE, --出勤日期(精确到日)
    time_range                VARCHAR2(2000), --资源日历出勤时间表达式
    work_datetime             DATE, --当前工作时间(完整时间)
    attendance_datetime_begin DATE, --出勤开始时间(精确到分钟)
    attendance_datetime_end   DATE, --当前工作时间(精确到分钟)
    datetime_range_arr        datetime_range_list --当前时间间隔
    );
    
--全局包变量：生命周期与数据库连接一致
  v_g_datetime_range_list datetime_range_list;

v_g_datetime_range_list                := str_to_datetime_range_list(rec.date_attendance
                                                                          ,rec.time_range);

  /*====================================================================================================
  用途说明：将包变量v_g_datetime_range_list转换为可用于表函数“table()”进行查询的程序！
  备注：此乃管道函数！用于将集合类型的全局包变量转换为可用于表函数“table()”进行查询的程序！
  ====================================================================================================*/
  FUNCTION datetime_range_list_to_table RETURN datetime_range_list
    PIPELINED IS
  BEGIN
    FOR i IN 1 .. v_g_datetime_range_list.count
    LOOP
      PIPE ROW(v_g_datetime_range_list(i));
    END LOOP;
  END;
  
  v_g_datetime_range_list                := str_to_datetime_range_list(rec.date_attendance
                                                                          ,rec.time_range);
  
  --使用
  SELECT MIN(datetime_begin)
              ,MAX(datetime_end)
          INTO v_attendance_info.attendance_datetime_begin
              ,v_attendance_info.attendance_datetime_end
          FROM TABLE(datetime_range_list_to_table)
  
  
  
--  ==================================================================
create type obj_target as object(  
ptype varchar2(30),
   pvalue number
);  
/

create or replace type typ_array_target as table of obj_target;  
/

CREATE OR REPLACE
function pipe_target(p_source_data in sys_refcursor)  
   return typ_array_target  
   pipelined  
   parallel_enable(partition p_source_data by any) is  

   r_target_data obj_target := obj_target(null, null);  

-- type typ_source_data is table of %rowtype index by pls_integer;    
   aa_source_data typ_array_target;  

begin    
   loop  
     fetch p_source_data bulk collect ------- 这里必须用BULK COLLECT; 还要加上LIMIT,否则外循环无意义
       into aa_source_data;  
    exit when aa_source_data.count = 0;      
     for i in 1 .. aa_source_data.count loop        
       -- r_target_data.pvalue       := aa_source_data(i).pvalue;  
       -- r_target_data.ptype := aa_source_data(i).ptype; 
       pipe row(aa_source_data(i));    ------- 原来写的是错误的。要么改用r_target_data但是没有必要
     end loop;      
   end loop;    
   close p_source_data;  
   return;  

end;  
/

create table test_pipetest (ttype varchar2(30),value number);
INSERT INTO test_pipetest VALUES(‘A‘,1);
INSERT INTO test_pipetest VALUES(‘B‘,2);

select * from table(pipe_target(cursor (select
                                  obj_target(s.ttype,sum(s.value)) pvalue ----- 你的SQL必须保证每行数据都是obj_target类型
                                    from test_pipetest s
                                    group by ttype  --------- 汇总完全没有问题
                                    )))

PTYPE                              PVALUE
------------------------------ ----------
A                                       1
B                                       2