13 Using Pipelined and Parallel Table Functions

Table Functions

Table functions return a collection type instance and can be queried like a table by calling the function in the FROM clause of a query. Table functions use the TABLE keyword.

The following example shows a table function GetBooks that takes a CLOB as input and returns an instance of the collection type BookSet_t. The CLOB column stores a catalog listing of books in some format (either proprietary or following a standard such as XML). The table function returns all the catalogs and their corresponding book listings. The collection type BookSet_t is defined in Example 13-1.

CREATE TYPE Book_t AS OBJECT
( name VARCHAR2(100),
  author VARCHAR2(30),
  abstract VARCHAR2(1000));

CREATE TYPE BookSet_t AS TABLE OF Book_t;

The CLOBs are stored in a table Catalogs, as demonstrated in Example 13-2.

CREATE TABLE Catalogs
( name VARCHAR2(30), 
  cat CLOB);

Function GetBooks() is defined in Example 13-3.

CREATE FUNCTION GetBooks(a CLOB) RETURN BookSet_t;

The query in Example 13-4 returns all the catalogs and their corresponding book listings.

SELECT c.name, Book.name, Book.author, Book.abstract
  FROM Catalogs c, TABLE(GetBooks(c.cat)) Book;

Pipelined Table Functions

Data is said to be pipelined if it is consumed by a consumer (transformation) as soon as the producer (transformation) produces it, without being staged in tables or a cache before being input to the next transformation.

Pipelining enables a table function to return rows faster and can reduce the memory required to cache a table function's results.

A pipelined table function can return the table function's result collection in subsets. The returned collection behaves like a stream that can be fetched from on demand. This makes it possible to use a table function like a virtual table.

Pipelined table functions can be implemented in two ways:

• In the native PL/SQL approach, the consumer and producers can run on separate execution threads (either in the same or different process context) and communicate through a pipe or queuing mechanism. This approach is similar to co-routine execution.

• In the interface approach, the consumer and producers run on the same execution thread. Producer explicitly returns the control back to the consumer after producing a set of results. In addition, the producer caches the current state so that it can resume where it left off when the consumer invokes it again.

  The interface approach requires you to implement a set of well-defined interfaces in a procedural language.

The co-routine execution model provides a simpler, native PL/SQL mechanism for implementing pipelined table functions, but this model cannot be used for table functions written in C or Java. The interface approach, on the other hand, can. The interface approach requires the producer to save the current state information in a context object before returning so that this state can be restored on the next invocation.

In the rest of this chapter, the term table function is used to refer to a pipelined table function— a table function that returns a collection in an iterative, pipelined way.

Pipelined Table Functions with REF CURSOR Arguments

A pipelined table function can accept any argument that regular functions accept. A table function that accepts a REF CURSOR as an argument can serve as a transformation function. That is, it can use the REF CURSOR to fetch the input rows, perform some transformation on them, and then pipeline the results out (using either the interface approach or the native PL/SQL approach).

For example, the following code sketches the declarations that define a StockPivot function. This function converts a row of the type (Ticker, OpenPrice, ClosePrice) into two rows of the form (Ticker, PriceType, Price). Calling StockPivot for the row ("ORCL", 41, 42) generates two rows: ("ORCL", "O", 41) and ("ORCL", "C", 42).

Input data for the table function might come from a source such as table StockTable:

CREATE TABLE StockTable (
  ticker VARCHAR(4),
  openprice NUMBER,
  closeprice NUMBER
);

The declarations are in Example 13-5.

-- Create the types for the table function's output collection 
-- and collection elements

CREATE TYPE TickerType AS OBJECT 
(
  ticker VARCHAR2(4),
  PriceType VARCHAR2(1),
  price NUMBER
);

CREATE TYPE TickerTypeSet AS TABLE OF TickerType;

-- Define the ref cursor type

CREATE PACKAGE refcur_pkg IS
  TYPE refcur_t IS REF CURSOR RETURN StockTable%ROWTYPE;
END refcur_pkg;
/

-- Create the table function

CREATE FUNCTION StockPivot(p refcur_pkg.refcur_t) RETURN TickerTypeSet
PIPELINED ... ;
/

Example 13-6 uses the StockPivot table function.

SELECT * FROM TABLE(StockPivot(CURSOR(SELECT * FROM StockTable)));

In the preceding query, the pipelined table function StockPivot fetches rows from the CURSOR subquery SELECT * FROM StockTable, performs the transformation, and pipelines the results back to the user as a table. The function produces two output rows (collection elements) for each input row.

Note that when a CURSOR subquery is passed from SQL to a REF CURSOR function argument as in the preceding example, the referenced cursor is open when the function begins executing.

Parallel Execution of Table Functions

With parallel execution of a function that appears in the SELECT list, execution of the function is pushed down to and conducted by multiple slave scan processes. These each execute the function on a segment of the function's input data.

For example, the query

SELECT f(col1) FROM tab;

is parallelized if f is a pure function. The SQL executed by a slave scan process is similar to:

SELECT f(col1) FROM tab WHERE ROWID BETWEEN :b1 AND :b2;

Each slave scan operates on a range of rowids and applies function f to each contained row. Function f is then executed by the scan processes; it does not run independently of them.

Unlike a function that appears in the SELECT list, a table function is called in the FROM clause and returns a collection. This affects the way that table function input data is partitioned among slave scans because the partitioning approach must be appropriate for the operation that the table function performs. (For example, an ORDER BY operation requires input to be range-partitioned, whereas a GROUP BY operation requires input to be hash partitioned.)

A table function itself specifies in its declaration the partitioning approach that is appropriate for it, as described in "Input Data Partitioning". The function is then executed in a two-stage operation. First, one set of slave processes partitions the data as directed in the function's declaration; then a second set of slave scans executes the table function in parallel on the partitioned data. The table function in the following query has a REF CURSOR parameter:

SELECT * FROM TABLE(f(CURSOR(SELECT * FROM tab)));

The scan is performed by one set of slave processes, which redistributes the rows (based on the partitioning method specified in the function declaration) to a second set of slave processes that actually executes function f in parallel.

Implementation Choices for Pipelined Table Functions

As noted previously, two approaches are supported for implementing pipelined table functions: the interface approach and the PL/SQL approach.

The interface approach requires the user to supply a type that implements a predefined Oracle interface consisting of start, fetch, and close operations. The type is associated with the table function when the table function is created. During query execution, the fetch method is invoked repeatedly to iteratively retrieve the results. With the interface approach, the methods of the implementation type associated with the table function can be implemented in any of the supported internal or external languages (including PL/SQL, C/C++, and Java).

With the PL/SQL approach, a single PL/SQL function includes a special instruction to pipeline results (single elements of the collection) out of the function instead of returning the whole collection as a single value. The native PL/SQL approach is simpler to implement because it requires writing only one PL/SQL function.

The approach used to implement pipelined table functions does not affect the way they are used. Pipelined table functions are used in SQL statements in exactly the same way regardless of the approach used to implement them.

Declarations of Pipelined Table Functions

You declare a pipelined table function by specifying the PIPELINED keyword. This keyword indicates that the function returns rows iteratively. The return type of the pipelined table function must be a collection type (a nested table or a varray).

Example 13-7 shows declarations of pipelined table functions implemented using the interface approach. The interface routines for functions GetBooks and StockPivot have been implemented in the types BookMethods and StockPivotImpl, respectively.

CREATE FUNCTION GetBooks(cat CLOB) RETURN BookSet_t PIPELINED USING BookMethods;

CREATE FUNCTION StockPivot(p refcur_pkg.refcur_t) 
  RETURN TickerTypeSet PIPELINED USING StockPivotImpl;

Example 13-8 shows declarations of the same table functions implemented using the native PL/SQL approach:

CREATE FUNCTION GetBooks(cat CLOB) RETURN BookSet_t PIPELINED IS ...;

CREATE FUNCTION StockPivot(p refcur_pkg.refcur_t) RETURN TickerTypeSet
PIPELINED IS...;

Implementing the Native PL/SQL Approach

In PL/SQL, the PIPE ROW statement causes a table function to pipe a row and continue processing. The statement enables a PL/SQL table function to return rows as soon as they are produced. This is demonstrated in Example 13-9. For performance reasons, the PL/SQL run-time system provides the rows to the consumer in batches.

CREATE FUNCTION StockPivot(p refcur_pkg.refcur_t) RETURN TickerTypeSet
PIPELINED IS
  out_rec TickerType := TickerType(NULL,NULL,NULL);
  in_rec p%ROWTYPE;
BEGIN
  LOOP
    FETCH p INTO in_rec; 
    EXIT WHEN p%NOTFOUND;
    -- first row
    out_rec.ticker := in_rec.Ticker;
    out_rec.PriceType := 'O';
    out_rec.price := in_rec.OpenPrice;
    PIPE ROW(out_rec);
    -- second row
    out_rec.PriceType := 'C';   
    out_rec.Price := in_rec.ClosePrice;
    PIPE ROW(out_rec);
  END LOOP;
  CLOSE p;
  RETURN;
END;
/

In Example 13-9, the PIPE ROW(out_rec) statement pipelines data out of the PL/SQL table function.

The PIPE ROW statement may be used only in the body of pipelined table functions; an error is raised if it is used anywhere else. The PIPE ROW statement can be omitted for a pipelined table function that returns no rows.

A pipelined table function must have a RETURN statement that does not return a value. The RETURN statement transfers the control back to the consumer and ensures that the next fetch gets a NO_DATA_FOUND exception.

Pipelining Between PL/SQL Table Functions

With serial execution, results are pipelined from one PL/SQL table function to another using an approach similar to co-routine execution. Example 13-10 pipelines results from function g to function f.

SELECT * FROM TABLE(f(CURSOR(SELECT * FROM TABLE(g()))));

Parallel execution works similarly, except that each function executes in a different process or set of processes.

Combining PIPE ROW with AUTONOMOUS_TRANSACTION

Because table functions pass control back and forth to a calling routine as rows are produced, there is a restriction on combining table functions and PRAGMA AUTONOMOUS_TRANSACTIONs. If a table function is part of an autonomous transaction, it must COMMIT or ROLLBACK before each PIPE ROW statement, to avoid an error in the calling subprogram.

Implementing the Interface Approach

To use the interface approach, you must define an implementation type that implements the ODCITable interface. This interface consists of start, fetch, and close routines whose signatures are specified by Oracle and which you implement as methods of the type.

Oracle invokes the methods to perform the following steps in the execution of a query that contains a table function:

1. Start by initializing the scan context parameter, using the ODCITableStart() function.

2. Fetch to produce a subset of the rows in the result collection. The ODCITableFetch() method is invoked as many times as necessary to return the entire collection.

3. Close and clean up (release memory and so on) using ODCITableClose() after the last ODCITableFetch().

The ODCITable interface also defines two optional routines, ODCITablePrepare() and ODCITableDescribe(), that are invoked at compilation time:

• ODCITableDescribe() determines the structure of the data type the table function returns, in situations where this cannot be defined in a static manner.

• ODCITablePrepare() initializes the scan context parameter. If this method is implemented, the scan context it prepares is passed to the ODCITableStart() routine, and the context is maintained between restarts of the table function. It also provides projection information and supports the return of transient anonymous types.

Close Routine

The close routine ODCITableClose() is invoked by Oracle after the last fetch invocation. The scan context is passed in as a parameter. This routine performs the necessary cleanup operations.

Figure 13-3 Flowchart of Table Function Row Source Execution

Description of "Figure 13-3 Flowchart of Table Function Row Source Execution"

CREATE FUNCTION AnyDocuments(VARCHAR2) RETURN ANYDATASET 
PIPELINED USING DocumentMethods;

SELECT * FROM 
  TABLE(AnyDocuments('http://.../documents.xml')) x
  WHERE x.Abstract like '%internet%';

CREATE TYPE Mag_t AS OBJECT
(   name VARCHAR2(100),
    publisher VARCHAR2(30),
    abstract VARCHAR2(1000)
);

STATIC FUNCTION ODCITableDescribe(rtype OUT ANYTYPE, 
                                        url VARCHAR2)
IS BEGIN    
    Contact specified web server and retrieve document...
    Check XML doc schema to determine if books or mags...
    IF books THEN
        rtype=AnyType.AnyTypeGetPersistent('SYS','BOOK_T');
    ELSE
        rtype=AnyType.AnyTypeGetPersistent('SYS','MAG_T');
    END IF;
END;

-- Issue a query for books
SELECT x.Name, x.Author
FROM TABLE(AnyDocuments('Books.xml')) x;

-- Issue a query for magazines
SELECT x.Name, x.Publisher
FROM TABLE(AnyDocuments('Magazines.xml')) x;

CREATE TYPE SYS.ODCITabFuncInto AS OBJECT (
  Attrs SYS.ODCINumberList,
  RetType SYS.AnyType
);

Querying Table Functions

Pipelined table functions are used in the FROM clause of SELECT statements independently from implementation, either in native PL/SQL or through the interface approach. The result rows are retrieved by Oracle iteratively from the table function implementation, as demonstrated in Example 13-15.

SELECT x.Ticker, x.Price 
FROM TABLE(StockPivot(CURSOR(SELECT * FROM StockTable))) x
WHERE x.PriceType='C';

SELECT * FROM TABLE(f(...)) t1, TABLE(f(...)) t2 
  WHERE t1.id = t2.id;
  
SELECT * FROM TABLE(f());

SELECT * FROM TABLE(f());

OPEN c FOR SELECT * FROM TABLE(f(...));

BEGIN
    OPEN r FOR SELECT * FROM TABLE(f(CURSOR(SELECT * FROM tab)));
    SELECT * BULK COLLECT INTO rec_tab FROM TABLE(g(r));
END;

SELECT * FROM TABLE(g(CURSOR(SELECT * FROM
  TABLE(f(CURSOR(SELECT * FROM tab))))));

Performing DML Operations Inside Table Functions

A table function must be declared with the autonomous transaction pragma in order for the function to execute DML statements. This pragma causes the function to execute in an autonomous transaction not shared by other processes, as demonstrated in Example 13-20.

CREATE FUNCTION f(p SYS_REFCURSOR) return CollType PIPELINED IS
    PRAGMA AUTONOMOUS_TRANSACTION; 
BEGIN ... END;

During parallel execution, each instance of the table function creates an independent transaction.

Performing DML Operations on Table Functions

Table functions cannot be the target table in UPDATE, INSERT, or DELETE statements. For example, the following statements raise an error:

UPDATE F(CURSOR(SELECT * FROM tab)) SET col = value; 
INSERT INTO f(...) VALUES ('any', 'thing'); 

However, you can create a view over a table function and use INSTEAD OF triggers to update it, as in Example 13-21.

CREATE VIEW BookTable AS 
  SELECT x.Name, x.Author
  FROM TABLE(GetBooks('data.txt')) x;

Example 13-22 demonstrates how an INSTEAD OF trigger is fired when the user inserts a row into the BookTable view:.

CREATE TRIGGER BookTable_insert
INSTEAD OF INSERT ON BookTable
REFERENCING NEW AS n
FOR EACH ROW
BEGIN
  ...
END;
INSERT INTO BookTable VALUES (...);

INSTEAD OF triggers can be defined for all DML operations on a view built on a table function.

Handling Exceptions in Table Functions

Exception handling in table functions works just as it does with ordinary user-defined functions.

Some languages, such as C and Java, provide a mechanism for user-supplied exception handling. If an exception raised within a table function is handled, the table function executes the exception handler and continues processing. Exiting the exception handler takes control to the enclosing scope. If the exception is cleared, execution proceeds normally.

An unhandled exception in a table function causes the parent transaction to roll back.

Inputting Data with Cursor Variables

You can pass a set of rows to a PL/SQL function in a REF CURSOR parameter, as demonstrated in Example 13-23.

FUNCTION f(p1 IN SYS_REFCURSOR) RETURN ... ;

Results of a subquery can be passed to a function directly, as demonstrated in Example 13-24. The CURSOR keyword is required to indicate that the results of a subquery should be passed as a REF CURSOR parameter.

SELECT * FROM TABLE(f(CURSOR(SELECT empno FROM tab)));

CREATE FUNCTION g(p1 pkg.refcur_t1, p2 pkg.refcur_t2) RETURN...
  PIPELINED ... ;

SELECT * FROM TABLE(g(CURSOR(SELECT empno FROM tab),
  CURSOR(SELECT * FROM emp));

SELECT * FROM TABLE(f(CURSOR(SELECT * FROM TABLE(g(...)))));

BEGIN
  OPEN r FOR SELECT * FROM TABLE(f(...));
  -- Must return a single row result set.
  SELECT * INTO rec FROM TABLE(g(r));
END;

CREATE OR replace PACKAGE p1 AS 
  TYPE rc IS REF cursor; 
  END; 

CREATE OR REPLACE LIBRARY MYLIB AS 'mylib.so'; 

CREATE OR REPLACE FUNCTION MyCallout (stmthp p1.rc) 
  RETURN binary_integer AS LANGUAGE C LIBRARY MYLIB 
  WITH CONTEXT 
  PARAMETERS (context, stmthp ocirefcursor, RETURN sb4); 

sb4 MyCallout (OCIExtProcContext *ctx, OCIStmt ** stmthp) 
  OCIEnv *envhp;                /* env. handle */ 
  OCISvcCtx *svchp;             /* service handle */ 
  OCIError *errhp;              /* error handle */ 
  OCISession *usrhp;            /* user handle */ 

  int errnum = 29400;           /* choose some oracle error number */ 
  char errmsg[512];             /* error message buffer */ 
  size_t errmsglen;             /* Length of error message */ 
  OCIDefine *defn1p = (OCIDefine *) 0; 
  OCINumber *val=(OCINumber *)0; 

  OCINumber *rval = (OCINumber *)0; 
  sword status =  0; 
  double num=0; 
  val = (OCINumber*) OCIExtProcAllocCallMemory(ctx, sizeof(OCINumber)); 
  /* Get OCI handles */ 
  if (GetHandles(ctx, &envhp, &svchp, &errhp, &usrhp,&rval)) 
    return -1; 
  /* Define the fetch buffer */ 
  psdro_checkerr(NULL, errhp, OCIDefineByPos(*stmthp, &defn1p, errhp, (ub4) 1, 
                                            (dvoid *) &num, (sb4) sizeof(num), 
                                            SQLT_FLT, (dvoid *) 0, (ub2 *)0, 
                                            (ub2 *)0, (ub4) OCI_DEFAULT)); 

  /* Fetch loop */ 
  while ((status = OCIStmtFetch(*stmthp, errhp, (ub4) 1,  (ub4) OCI_FETCH_NEXT, 
                                (ub4) OCI_DEFAULT)) == OCI_SUCCESS || 
         status == OCI_SUCCESS_WITH_INFO) 
  { 
    printf("val=%lf\n",num); 
  } 
  return 0; 
} 

Input Data Partitioning

The table function declaration can specify data partitioning for exactly one REF CURSOR parameter, as demonstrated in Example 13-30. The PARTITION BY phrase in the PARALLEL_ENABLE clause specifies which one of the input cursors to partition, and what columns to use for partitioning.

CREATE FUNCTION f(p ref_cursor_type) RETURN rec_tab_type PIPELINED
  PARALLEL_ENABLE(PARTITION p BY [{HASH | RANGE} (column_list) | ANY ]) IS
BEGIN ... END;

When explicit column names are specified in the column list, the partitioning method may be either RANGE or HASH. The input rows are hash- or range-partitioned on the specified columns.

The ANY keyword enables you to indicate that the function behavior is independent of the partitioning of the input data. When this keyword is used, the run-time system randomly partitions the data among the slaves. This keyword is appropriate for use with functions that take in one row, manipulate its columns, and generate output row(s) based on the columns of this row only.

To demonstrate, the pivot-like function StockPivot() in Example 13-31 takes as input a row of the type (Ticker varchar(4), OpenPrice number, ClosePrice number), and generates rows of the type (Ticker varchar(4), PriceType varchar(1), Price number). Thus, the row ("ORCL", 41, 42) generates two rows: ("ORCL", "O", 41) and ("ORCL", "C", 42).

You can use directly TickerTypeSet, as created in Example 13-5, instead of rec_tab_type or define it the same way.

CREATE FUNCTION StockPivot(p refcur_pkg.refcur_t) RETURN rec_tab_type PIPELINED
   PARALLEL_ENABLE(PARTITION p BY ANY) IS
   ret_rec rec_type;
 
BEGIN
   FOR rec IN p LOOP
      ret_rec.Ticker := rec.Ticker;
      ret_rec.PriceType := "O";
      ret_rec.Price := rec.OpenPrice;
      PIPE ROW(ret_rec);
 
      ret_rec.Ticker := rec.Ticker;   -- Redundant; not required
      ret_rec.PriceType := "C";
      ret_rec.Price := rec.ClosePrice;
      PIPE ROW ret_rec;
   END LOOP;
   RETURN;
 
END;

The function f() may be used to generate another table from Stocks table, as shown in Example 13-32.

INSERT INTO AlternateStockTable
   SELECT * FROM 
   TABLE(StockPivot(CURSOR(SELECT * FROM StockTable)));

If StockTable is scanned in parallel and partitioned on OpenPrice, then the function StockPivot() is combined with the data-flow operator that scans StockTable and therefore sees the same partitioning.

If StockTable is not partitioned, and the scan on it does not execute in parallel, the insert into AlternateStockTable also runs sequentially, as demonstrated in Example 13-33.

CREATE FUNCTION g(p refcur_pkg.refcur_t) RETURN ... PIPELINED
  PARALLEL_ENABLE (PARTITION p BY ANY)
BEGIN 
  ... 
END;

INSERT INTO AlternateStockTable
  SELECT * FROM TABLE(f(CURSOR(SELECT * FROM Stocks))), TABLE(g(CURSOR( ... )))
    WHERE join_condition;

If function g() runs in parallel and is partitioned by ANY, then the parallel insert can belong in the same data-flow operator as g().

Whenever the ANY keyword is specified, the data is partitioned randomly among the slaves. This effectively means that the function is executed in the same slave set which does the scan associated with the input parameter.

No redistribution or repartitioning of the data is required here. In the case, when the cursor p itself is not parallel-enabled, the incoming data is randomly partitioned on the columns in the column list. The round-robin table queue is used for this partitioning.

Parallel Execution of Leaf-Level Table Functions

Parallel execution of a table function requires a REF CURSOR as an input parameter. If the table function does not inherently require a set of rows as input, then you must redesign the function to require a REF CURSOR as input.

For example, the following function reads a set of external files in parallel and returns the records they contain. To provide work for a REF CURSOR, you might first create a table and populate it with the filenames. A REF CURSOR over this table can then be passed as a parameter to the table function readfiles(), as demonstrated by Example 13-34.

CREATE TABLE filetab(filename VARCHAR(20));

INSERT INTO filetab VALUES('file0');   
INSERT INTO filetab VALUES('file1');  
...
INSERT INTO filetab VALUES('fileN');

SELECT * FROM TABLE(readfiles(CURSOR(SELECT filename FROM filetab)));

CREATE FUNCTION readfiles(p pkg.rc_t) RETURN coll_type
  PARALLEL_ENABLE(PARTITION p BY ANY) IS
  ret_rec rec_type;
BEGIN
  FOR rec IN p LOOP
    done := FALSE;
    WHILE (done = FALSE) LOOP
         done := readfilerecord(rec.filename, ret_rec);
         PIPE ROW(ret_rec);
    END LOOP;
  END LOOP;
  RETURN;
END;

Parallel Execution: Partitioning and Clustering

Partitioning and clustering are easily confused, but they do different things. Sometimes partitioning can be sufficient without clustering in parallel execution.

Consider a function SmallAggr that performs in-memory aggregation of salary for each department_id, where department_id can be either 1, 2, or 3. The input rows to the function can be partitioned by HASH on department_id so that all rows with department_id equal to 1 go to one slave, all rows with department_id equal to 2 go to another slave, and so on.

The input rows do not have to be clustered on department_id to perform the aggregation in the function. Each slave could have a 1 by 3 array SmallSum[1..3], in which the aggregate sum for each department_id is added in memory into SmallSum[department_id]. On the other hand, if the number of unique values of department_id were very large, you would want to use clustering to compute department aggregates and write them to disk one department_id at a time.