/*
DROP TABLE "TEST_INPUT";
DROP TABLE "TEST_ITEM";
DROP TABLE "TEST_NODE";
DROP SEQUENCE "TEST_ITEM_SEQ";
DROP SEQUENCE "TEST_NODE_SEQ";
DROP PACKAGE "TEST_PIPELINE";
DROP PACKAGE BODY "TEST_PIPELINE";
DROP TYPE "TEST_NODE_OT";
DROP TYPE "TEST_ITEM_OT";
DROP TYPE "TEST_NTT";
DROP TYPE "TEST_OT";
--*/

--------------------------------------------------------
--  DDL for Type TEST_OT
--------------------------------------------------------

  CREATE OR REPLACE EDITIONABLE TYPE "TEST_OT" FORCE AS OBJECT  
( 
  TEST_SID VARCHAR2(50 CHAR)
) NOT FINAL;

/
--------------------------------------------------------
--  DDL for Type TEST_NTT
--------------------------------------------------------

  CREATE OR REPLACE EDITIONABLE TYPE "TEST_NTT" AS TABLE OF TEST_OT;

/
--------------------------------------------------------
--  DDL for Type TEST_NODE_OT
--------------------------------------------------------

  CREATE OR REPLACE EDITIONABLE TYPE "TEST_NODE_OT" UNDER TEST_OT
( 
  NODE_ID    NUMBER,
	NODE_NAME  VARCHAR2(50 CHAR), 
  CONSTRUCTOR FUNCTION TEST_NODE_OT (p_node_id number, p_node_type varchar2) RETURN SELF AS RESULT
) FINAL;
/
CREATE OR REPLACE EDITIONABLE TYPE BODY "TEST_NODE_OT" AS
  CONSTRUCTOR FUNCTION TEST_NODE_OT(p_node_id number, p_node_type varchar2)
       RETURN SELF AS RESULT IS
        l_sid number;
     BEGIN
        l_sid := sys_context('USERENV', 'SID');
    
        NODE_ID         :=  p_node_id;
        NODE_NAME       :=  'Node: '||p_node_type;
        TEST_SID        :=  'thread: '||l_sid;

        RETURN;
     END;
END;

/
--------------------------------------------------------
--  DDL for Type TEST_ITEM_OT
--------------------------------------------------------

  CREATE OR REPLACE EDITIONABLE TYPE "TEST_ITEM_OT" UNDER TEST_OT
( 
  ITEM_ID    NUMBER,
  ITEM_TYPE  VARCHAR2(50),
	ITEM_NAME  VARCHAR2(50 CHAR), 
  NODE_ID    NUMBER,
  PARENT_ID  NUMBER,
  CONSTRUCTOR FUNCTION TEST_ITEM_OT (p_item_id number, p_item_type varchar2, p_node_id number, p_parent_id number) RETURN SELF AS RESULT
) FINAL;
/
CREATE OR REPLACE EDITIONABLE TYPE BODY "TEST_ITEM_OT" AS
  CONSTRUCTOR FUNCTION TEST_ITEM_OT(p_item_id number, p_item_type varchar2, p_node_id number, p_parent_id number)
       RETURN SELF AS RESULT IS
        l_sid number;
     BEGIN
        l_sid := sys_context('USERENV', 'SID');
    
        ITEM_ID     :=  p_item_id;
        ITEM_TYPE   :=  p_item_type;
        ITEM_NAME   :=  p_item_type||'_'||to_char(ITEM_ID);
        NODE_ID     :=  p_node_id;
        PARENT_ID   :=  p_parent_id;  
        TEST_SID    :=  'thread: '||l_sid;

        RETURN;
     END;
END;

/
--------------------------------------------------------
--  DDL for Sequence TEST_ITEM_SEQ
--------------------------------------------------------

   CREATE SEQUENCE  "TEST_ITEM_SEQ"  MINVALUE 1 MAXVALUE 99999999999999 INCREMENT BY 1 START WITH 35529301 CACHE 100 NOORDER  NOCYCLE  NOPARTITION ;
--------------------------------------------------------
--  DDL for Sequence TEST_NODE_SEQ
--------------------------------------------------------

   CREATE SEQUENCE  "TEST_NODE_SEQ"  MINVALUE 1 MAXVALUE 99999999999999 INCREMENT BY 1 START WITH 1362601 CACHE 100 NOORDER  NOCYCLE  NOPARTITION ;
--------------------------------------------------------
--  DDL for Table TEST_INPUT
--------------------------------------------------------

  CREATE TABLE "TEST_INPUT" 
   (	"NODE_TYPE" VARCHAR2(50 CHAR), 
	"ITEM_TYPE" VARCHAR2(50 CHAR), 
	"ITEM_COUNT" NUMBER
   ) ;
--------------------------------------------------------
--  DDL for Table TEST_ITEM
--------------------------------------------------------

  CREATE TABLE "TEST_ITEM" 
   (	"ITEM_ID" NUMBER, 
	"ITEM_TYPE" VARCHAR2(50), 
	"ITEM_NAME" VARCHAR2(50 CHAR), 
	"NODE_ID" NUMBER, 
	"PARENT_ID" NUMBER, 
	"TEST_SID" VARCHAR2(50 CHAR)
   ) ;
--------------------------------------------------------
--  DDL for Table TEST_NODE
--------------------------------------------------------

  CREATE TABLE "TEST_NODE" 
   (	"NODE_ID" NUMBER, 
	"NODE_NAME" VARCHAR2(50 CHAR), 
	"TEST_SID" VARCHAR2(50 CHAR)
   ) ;
--------------------------------------------------------
--  DDL for Package TEST_PIPELINE
--------------------------------------------------------

  CREATE OR REPLACE EDITIONABLE PACKAGE "TEST_PIPELINE" AS 

TYPE t_test_input IS REF CURSOR RETURN TEST_INPUT%rowtype;

FUNCTION f_test (p_cursor IN t_test_input, p_limit_size IN PLS_INTEGER DEFAULT 1000)
  RETURN test_ntt PIPELINED
  CLUSTER p_cursor BY (node_type) 
  PARALLEL_ENABLE  
    (PARTITION p_cursor BY HASH(node_type))
--    (PARTITION p_cursor BY ANY);
  ;
END TEST_PIPELINE;

/
--------------------------------------------------------
--  DDL for Package Body TEST_PIPELINE
--------------------------------------------------------

  CREATE OR REPLACE EDITIONABLE PACKAGE BODY "TEST_PIPELINE" AS 


FUNCTION f_test (p_cursor IN t_test_input, p_limit_size IN PLS_INTEGER DEFAULT 1000)
  RETURN test_ntt PIPELINED
  CLUSTER p_cursor BY (node_type) 
  PARALLEL_ENABLE  
    (PARTITION p_cursor BY HASH(node_type))
--    (PARTITION p_cursor BY ANY)
  IS
  TYPE cursor_ntt IS TABLE OF TEST_INPUT%rowtype;
  nt_src_data cursor_ntt;
  l_node_id  NUMBER;
  l_item_id  NUMBER;
BEGIN
  LOOP
    FETCH p_cursor BULK COLLECT INTO nt_src_data LIMIT p_limit_size;
    EXIT WHEN nt_src_data.COUNT = 0;
    
    FOR i IN 1 .. nt_src_data.COUNT LOOP

      -- generate TEST_NODE row    
      l_node_id    :=  test_node_seq.NEXTVAL;
      PIPE ROW (TEST_NODE_OT (
                  p_node_id    =>    l_node_id, 
                  p_node_type  =>    nt_src_data(i).node_type
                ));
      
      FOR k in 1..nt_src_data(i).item_count LOOP
      
        -- generate TEST_ITEM row
        l_item_id    :=  test_item_seq.NEXTVAL;
        PIPE ROW (TEST_ITEM_OT (
                   p_item_id    =>  l_item_id,
                   p_item_type  =>  nt_src_data(i).item_type,
                   p_node_id    =>  l_node_id,
                   p_parent_id  =>  null
                  ));
        
        IF nt_src_data(i).item_type = 'item_item' THEN          
          PIPE ROW (TEST_ITEM_OT (
                     p_item_id    =>  test_item_seq.NEXTVAL,
                     p_item_type  =>  nt_src_data(i).item_type,
                     p_node_id    =>  l_node_id,
                     p_parent_id  =>  l_item_id
                    ));
        END IF;
        
      END LOOP; -- item_count
      
    END LOOP;  -- nt_src_data

  END LOOP; -- p_cursor
  
  CLOSE p_cursor;
  RETURN;
  
EXCEPTION
  WHEN NO_DATA_NEEDED THEN
    RETURN;
  WHEN OTHERS THEN
--    logEvent(l_module_name||'_FAILED', substr(dbms_utility.format_error_stack || ' >> ' || dbms_utility.format_error_backtrace, 1, 3000));
    raise;   
END f_test;

END TEST_PIPELINE;

/

-- generate sample data
INSERT /*+ PARALLEL(4) ENABLE_PARALLEL_DML */ ALL 
INTO test_input (node_type, item_type, item_count) VALUES ('node_'||test_id, 'item1', 100 )
INTO test_input (node_type, item_type, item_count) VALUES ('node_'||test_id, 'item2', 200 )
INTO test_input (node_type, item_type, item_count) VALUES ('node_'||test_id, 'item_item', 10 )
WITH p AS (SELECT ROWNUM as test_id FROM DUAL CONNECT BY LEVEL <= 100)
SELECT * FROM p
;