本文共 13014 字，大约阅读时间需要 43 分钟。

说明：本文为面向PL/SQL开发初学者的指导手册

标签：管道函数、管道表函数、表函数、管道、pipe

易学：文中删去了不需要的多余部分，让初学者一目了然一学就会

温馨提示：如果您发现本文哪里写的有问题或者有更好的写法请留言或私信我进行修改优化

---

★ 名词解释

表函数容许在SQL中调用PL/SQL的功能 管道函数又叫管道表函数也是一种表函数，它的运行特点是在函数运行期间“即时的”返回PL/SQL集合数据

★ 案例 SQL> INSERT INTO   tickertable (ticker, price_type, price, price_date)             SELECT ticker, price_type, price, price_date             FROM   TABLE(stockpivot_pkg.pipe_stocks(CURSOR(SELECT * FROM stocktable)));

★ 工作流程 ********************************* *                  ·····   * *     SQL  管道函数  ·PL/SQL·   * *                  ·····   * *********************************

★ 相关文章 《Oracle PL/SQL程序设计 第6版》 ★ 场景说明 从外部表（stocktable）中提取数据，然后插入数据库表（tickertable）中

★ 脚本说明 ※ 环境准备：stockpivot_setup.sql ※ 执行方式：stockpivot_run.sql ※ 测试脚本：stockpivot_test.sql ※ 测试结果：stockpivot_test.txt ※ 清理脚本：stockpivot_teardown.sql

★ 测试内容 常规的PL/SQL 优化的PL/SQL 只用管道函数的PL/SQL 使用了bulck collect优化的管道函数的PL/SQL 使用了bulck collect+并行优化的管道函数的PL/SQL

★ 主要对象 ※ load_stocks_legacy   类型：存储过程   功能：用来一条一条的执行常规的DML   耗时：57.43 seconds   redo：273,003K ※ load_stocks_forall   类型：存储过程   功能：用来测试Bukck+forall的性能   耗时：5.07 seconds   redo：37,918K ※ load_stocks   类型：存储过程   功能：用来调用普通的管道函数（pipe_stocks）   耗时：16.47 seconds   redo：37,559K ※ load_stocks_array   类型：存储过程   功能：用来调用使用了数组的管道函数（pipe_stocks_array）   耗时：6.26 seconds   redo：37,550K ※ load_stocks_parallel   类型：存储过程   功能：用来调用使用了并行+数组的管道函数（pipe_stocks_parallel）   耗时：3.41 seconds   redo：25K

★ 相关信息 ※ 表 CREATE TABLE stocktable（外部表） CREATE TABLE tickertable（堆表） ※ SQL数据类型 CREATE TYPE stockpivot_ot  AS OBJECT（对象类型） CREATE TYPE stockpivot_ntt AS TABLE OF stockpivot_ot（集合） ※ PL/SQL数据类型 TYPE    stocktable_rct  IS REF CURSOR RETURN stocktable%ROWTYPE;（强类型的引用游标） TYPE    stocktable_aat  IS TABLE OF stocktable%ROWTYPE INDEX BY PLS_INTEGER;（集合） SUBTYPE stocktable_rt   IS stocktable%ROWTYPE;（游标） TYPE    tickertable_aat IS TABLE OF tickertable%ROWTYPE INDEX BY PLS_INTEGER;（集合） SUBTYPE tickertable_rt  IS tickertable%ROWTYPE;（游标）

 

★ 实验数据（stockpivot_setup.sql）

CREATE OR REPLACE DIRECTORY dir AS '/home/oracle';grant read,write on directory dir to scott;grant dba to scott;DECLARE   f UTL_FILE.FILE_TYPE := UTL_FILE.FOPEN('DIR', 'stocktable.dat', 'w');BEGIN   FOR r IN (WITH opening_prices AS (                     SELECT 'STK' || TO_CHAR(ROWNUM) AS ticker                     ,      ROUND(DBMS_RANDOM.VALUE(0, 2000), 4) AS open_price                     ,      SYSDATE-ABS(DBMS_RANDOM.VALUE(0,30)) AS trade_date                     FROM   dual                     CONNECT BY ROWNUM <= &num_rows                     )             SELECT ticker              ,      open_price              ,      ROUND(open_price * ABS(DBMS_RANDOM.VALUE(0.1,2)),4) AS close_price             ,      trade_date             FROM   opening_prices)   LOOP      UTL_FILE.PUT_LINE(f, r.ticker || ',' || r.open_price || ',' ||                            r.close_price || ',' || TO_CHAR(r.trade_date, 'DD/MM/YYYY'));   END LOOP;   UTL_FILE.FCLOSE(f);END;/CREATE TABLE stocktable( ticker      VARCHAR2(10), open_price  NUMBER, close_price NUMBER , trade_date  DATE)ORGANIZATION EXTERNAL(  TYPE ORACLE_LOADER  DEFAULT DIRECTORY dir  ACCESS PARAMETERS  (     RECORDS DELIMITED by NEWLINE     NOBADFILE     NOLOGFILE     NODISCARDFILE     FIELDS TERMINATED BY ','     ( ticker     , open_price     , close_price     , trade_date CHAR(20) DATE_FORMAT DATE MASK "DD/MM/YYYY"     )  )  LOCATION ('stocktable.dat'))REJECT LIMIT UNLIMITED;exec DBMS_STATS.GATHER_TABLE_STATS(USER, 'STOCKTABLE', estimate_percent=>NULL);CREATE TABLE tickertable( ticker      VARCHAR2(10), price_type  VARCHAR2(1), price       NUMBER , price_date  DATE);CREATE TYPE stockpivot_ot AS OBJECT( ticker      VARCHAR2(10), price_type  VARCHAR2(1), price       NUMBER , price_date  DATE);/CREATE TYPE stockpivot_ntt AS TABLE OF stockpivot_ot;/CREATE PACKAGE stockpivot_pkg AS   c_default_limit CONSTANT PLS_INTEGER := 100;   TYPE stocktable_rct IS REF CURSOR      RETURN stocktable%ROWTYPE;   TYPE stocktable_aat IS TABLE OF stocktable%ROWTYPE      INDEX BY PLS_INTEGER;   SUBTYPE stocktable_rt IS stocktable%ROWTYPE;   TYPE tickertable_aat IS TABLE OF tickertable%ROWTYPE      INDEX BY PLS_INTEGER;   SUBTYPE tickertable_rt IS tickertable%ROWTYPE;   PROCEDURE load_stocks_legacy;   PROCEDURE load_stocks_forall(             p_limit_size  IN PLS_INTEGER DEFAULT stockpivot_pkg.c_default_limit             );   FUNCTION pipe_stocks(             p_source_data IN stockpivot_pkg.stocktable_rct             ) RETURN stockpivot_ntt PIPELINED;   FUNCTION pipe_stocks_array(             p_source_data IN stockpivot_pkg.stocktable_rct,            p_limit_size  IN PLS_INTEGER DEFAULT stockpivot_pkg.c_default_limit            ) RETURN stockpivot_ntt PIPELINED;   FUNCTION pipe_stocks_parallel(             p_source_data IN stockpivot_pkg.stocktable_rct,            p_limit_size  IN PLS_INTEGER DEFAULT stockpivot_pkg.c_default_limit            ) RETURN stockpivot_ntt               PIPELINED              PARALLEL_ENABLE (PARTITION p_source_data BY ANY);   PROCEDURE load_stocks;   PROCEDURE load_stocks_array;   PROCEDURE load_stocks_parallel;END stockpivot_pkg;/CREATE PACKAGE BODY stockpivot_pkg AS   -----------------------------------------------------------------------   PROCEDURE load_stocks_legacy IS      CURSOR c_source_data IS         SELECT ticker, open_price, close_price, trade_date         FROM   stocktable;      r_source_data stockpivot_pkg.stocktable_rt;      r_target_data stockpivot_pkg.tickertable_rt;   BEGIN      OPEN c_source_data;      LOOP         FETCH c_source_data INTO r_source_data;         EXIT WHEN c_source_data%NOTFOUND;         /* Opening price... */         r_target_data.ticker      := r_source_data.ticker;         r_target_data.price_type  := 'O';         r_target_data.price       := r_source_data.open_price;         r_target_data.price_date  := r_source_data.trade_date;               INSERT INTO tickertable VALUES r_target_data;         /* Closing price... */         r_target_data.price_type := 'C';         r_target_data.price      := r_source_data.close_price;                  INSERT INTO tickertable VALUES r_target_data;      END LOOP;      DBMS_OUTPUT.PUT_LINE(          c_source_data%ROWCOUNT * 2 || ' rows inserted.' );      CLOSE c_source_data;   END load_stocks_legacy;   -----------------------------------------------------------------------   FUNCTION pipe_stocks(             p_source_data IN stockpivot_pkg.stocktable_rct             ) RETURN stockpivot_ntt PIPELINED IS      r_target_data stockpivot_ot := stockpivot_ot(NULL, NULL, NULL, NULL);      r_source_data stockpivot_pkg.stocktable_rt;   BEGIN      LOOP         FETCH p_source_data INTO r_source_data;         EXIT WHEN p_source_data%NOTFOUND;         /* First row... */         r_target_data.ticker     := r_source_data.ticker;         r_target_data.price_type := 'O';         r_target_data.price      := r_source_data.open_price;         r_target_data.price_date := r_source_data.trade_date;         PIPE ROW (r_target_data);         /* Second row... */         r_target_data.price_type := 'C';         r_target_data.price      := r_source_data.close_price;         PIPE ROW (r_target_data);      END LOOP;      CLOSE p_source_data;      RETURN;   END pipe_stocks;   -----------------------------------------------------------------------   FUNCTION pipe_stocks_array(             p_source_data IN stockpivot_pkg.stocktable_rct,            p_limit_size  IN PLS_INTEGER DEFAULT stockpivot_pkg.c_default_limit            ) RETURN stockpivot_ntt PIPELINED IS      r_target_data  stockpivot_ot := stockpivot_ot(NULL, NULL, NULL, NULL);      aa_source_data stockpivot_pkg.stocktable_aat;   BEGIN      LOOP         FETCH p_source_data BULK COLLECT INTO aa_source_data LIMIT p_limit_size;         EXIT WHEN aa_source_data.COUNT = 0;         /* Process the batch of (p_limit_size) records... */         FOR i IN 1 .. aa_source_data.COUNT LOOP               /* First row... */            r_target_data.ticker     := aa_source_data(i).ticker;            r_target_data.price_type := 'O';            r_target_data.price      := aa_source_data(i).open_price;            r_target_data.price_date := aa_source_data(i).trade_date;            PIPE ROW (r_target_data);            /* Second row... */            r_target_data.price_type := 'C';            r_target_data.price      := aa_source_data(i).close_price;            PIPE ROW (r_target_data);         END LOOP;      END LOOP;      CLOSE p_source_data;       RETURN;   END pipe_stocks_array;   -----------------------------------------------------------------------   FUNCTION pipe_stocks_parallel(             p_source_data IN stockpivot_pkg.stocktable_rct,            p_limit_size  IN PLS_INTEGER DEFAULT stockpivot_pkg.c_default_limit            ) RETURN stockpivot_ntt               PIPELINED              PARALLEL_ENABLE (PARTITION p_source_data BY ANY) IS      r_target_data  stockpivot_ot := stockpivot_ot(NULL, NULL, NULL, NULL);      aa_source_data stockpivot_pkg.stocktable_aat;   BEGIN      LOOP         FETCH p_source_data BULK COLLECT INTO aa_source_data LIMIT p_limit_size;         EXIT WHEN aa_source_data.COUNT = 0;         /* Process the batch of (p_limit_size) records... */         FOR i IN 1 .. aa_source_data.COUNT LOOP               /* First row... */            r_target_data.ticker     := aa_source_data(i).ticker;            r_target_data.price_type := 'O';            r_target_data.price      := aa_source_data(i).open_price;            r_target_data.price_date := aa_source_data(i).trade_date;            PIPE ROW (r_target_data);            /* Second row... */            r_target_data.price_type := 'C';            r_target_data.price      := aa_source_data(i).close_price;            PIPE ROW (r_target_data);         END LOOP;      END LOOP;      CLOSE p_source_data;       RETURN;   END pipe_stocks_parallel;   -----------------------------------------------------------------------   PROCEDURE load_stocks IS   BEGIN      INSERT INTO tickertable (ticker, price_type, price, price_date)      SELECT ticker, price_type, price, price_date      FROM   TABLE(                stockpivot_pkg.pipe_stocks(                   CURSOR(SELECT * FROM stocktable)));      DBMS_OUTPUT.PUT_LINE( SQL%ROWCOUNT || ' rows inserted.' );   END load_stocks;   -----------------------------------------------------------------------   PROCEDURE load_stocks_array IS   BEGIN      INSERT INTO tickertable (ticker, price_type, price, price_date)      SELECT ticker, price_type, price, price_date      FROM   TABLE(                stockpivot_pkg.pipe_stocks_array(                   CURSOR(SELECT * FROM stocktable)));      DBMS_OUTPUT.PUT_LINE( SQL%ROWCOUNT || ' rows inserted.' );   END load_stocks_array;   -----------------------------------------------------------------------   PROCEDURE load_stocks_parallel IS   BEGIN      EXECUTE IMMEDIATE 'ALTER SESSION ENABLE PARALLEL DML';      INSERT /*+ PARALLEL(t, 4) */ INTO tickertable t            (ticker, price_type, price, price_date)      SELECT ticker, price_type, price, price_date      FROM   TABLE(                stockpivot_pkg.pipe_stocks_parallel(                   CURSOR(SELECT /*+ PARALLEL(s, 4) */ * FROM stocktable s)));      DBMS_OUTPUT.PUT_LINE( SQL%ROWCOUNT || ' rows inserted.' );   END load_stocks_parallel;   -----------------------------------------------------------------------   PROCEDURE load_stocks_forall(             p_limit_size  IN PLS_INTEGER DEFAULT stockpivot_pkg.c_default_limit             ) IS      CURSOR c_source_data IS         SELECT ticker, open_price, close_price, trade_date         FROM   stocktable;      aa_source_data stockpivot_pkg.stocktable_aat;      aa_target_data stockpivot_pkg.tickertable_aat;      v_indx         PLS_INTEGER;      v_rowcount     PLS_INTEGER := 0;   BEGIN      OPEN c_source_data;      LOOP         FETCH c_source_data BULK COLLECT INTO aa_source_data LIMIT p_limit_size;         EXIT WHEN aa_source_data.COUNT = 0;         aa_target_data.DELETE;         FOR i IN 1 .. aa_source_data.COUNT LOOP            /* Opening price... */            v_indx := aa_target_data.COUNT + 1;            aa_target_data(v_indx).ticker      := aa_source_data(i).ticker;            aa_target_data(v_indx).price_type  := 'O';            aa_target_data(v_indx).price       := aa_source_data(i).open_price;            aa_target_data(v_indx).price_date  := aa_source_data(i).trade_date;                  /* Closing price... */            v_indx := aa_target_data.COUNT + 1;            aa_target_data(v_indx).ticker      := aa_source_data(i).ticker;            aa_target_data(v_indx).price_type  := 'C';            aa_target_data(v_indx).price       := aa_source_data(i).close_price;            aa_target_data(v_indx).price_date  := aa_source_data(i).trade_date;                  END LOOP;         FORALL i IN INDICES OF aa_target_data            INSERT INTO tickertable             VALUES aa_target_data(i);         v_rowcount := v_rowcount + SQL%ROWCOUNT;      END LOOP;      DBMS_OUTPUT.PUT_LINE( v_rowcount || ' rows inserted.' );      CLOSE c_source_data;   END load_stocks_forall;END stockpivot_pkg;/

 

---

※ 如果您觉得文章写的还不错， 别忘了在文末给作者点个赞哦 ~

over

转载地址：https://blog.csdn.net/zzt_2009/article/details/114396792 如侵犯您的版权，请留言回复原文章的地址，我们会给您删除此文章，给您带来不便请您谅解！