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.

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

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

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

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

CREATE TYPE BookSet_t AS TABLE OF Book_t;

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

CREATE FUNCTION GetBooks(a CLOB) RETURN BookSet_t;

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

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 this discussion, the term table function refers to a pipelined table function— a table function that returns a collection in an iterative, pipelined way.

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.

Example 13-6 uses the StockPivot table function.

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.

-- 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 ... ;
/

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

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.

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.

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.

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 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.

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.

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

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.

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.

If you create an XMLIndex index on a partitioned XMLType table, or a partitioned table with an XMLType column using range, list, or hash partitioning, that index is equipartitioned with the base table.

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;

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;

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

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;

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;

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.