创建自定义的多参数 Oracle 分析函数

Question

背景

我知道在 Oracle 中可以创建 custom aggregate functions 来处理一组值并返回一个结果。 编辑：我什至阅读了 docs.oracle.com/cd/B28359_01/appdev.111/b28425/aggr_functions.htm 上的友好手册！

我也知道 Oracle 提供 built-in analytic functions，如 DENSE_RANK 和 RATIO_TO_REPORT，它们为每个输入提供值，相对于输入所在的值的集合/窗口。

问题

我想知道是否有一种方法可以创建我自己的分析函数，大​​概与我可以创建自己的聚合函数类似的方式，特别是创建一个 我的自定义分析函数中的其他参数。

微妙的术语警告

当我提到“分析函数”时，请将其理解为一个函数，除了通过PARTITION 关键字接受窗口参数外，还可以在给定窗口内返回不同的值。 （如果有人对此有更好的说法，请告诉我！纯解析函数？DENSE_RANK-class 解析函数？非聚合解析函数？）

Oracle 文档指出聚合函数可以用作分析（窗口）函数。不幸的是，这仅意味着用于在分析函数中指定窗口的PARTITION 关键字也可以应用于聚合函数。它不会将聚合函数提升到我梦寐以求的状态，即能够在固定窗口内返回不同的值。

聚合用作分析：

SELECT SUM(income) OVER (PARTITION BY first_initial) AS total FROM data;

将有与data 一样多的记录，但它的不同totals 的数量与第一个首字母一样多。

解析用作解析：

SELECT RATIO_TO_REPORT(income) OVER (PARTITION BY first_initial) AS ratio FROM data;

将拥有与data 一样多的记录，并且即使在给定的first_initial 分区内，那些ratios 也可能都是不同的。

上下文

我已获得对 PL/SQL 过程的仅调用访问权限，该过程接受数字集合作为 IN OUT 参数，并具有一些其他 IN 配置参数。该程序以受配置参数影响的方式修改集合的值（将其想象为“大学专有的认可和要求的成绩曲线程序”）。

目前，使用该过程的过程是硬编码一个游标循环，该循环检测从一个数据分区到另一个数据分区的变化，然后在每个分区内将数据提取到一个集合中，然后将其传递给该过程、更改并最终转储到一个单独的表中。我计划通过创建一个封装了一些逻辑的PIPELINEDPARALLEL_ENABLE 表函数来改进这一点，但我更喜欢启用如下查询：

SELECT G.Course_ID
     , G.Student_ID
     , G.Raw_Grade
     , analytic_wrapper(G.raw_grade, P.course_config_data)
                  OVER (PARTITION BY G.Course_ID) AS Adjusted_Grade
     , P.course_config_data
  FROM      grades   G
  LEFT JOIN policies P
  ON G.Course_ID = P.Course_ID;

但是，这需要能够创建自定义分析函数，并且由于过程需要在不同分区上进行不同输入的方式（例如上面的Course_ID-特定P.course_config_data），它还必须接受不仅data-to-be-aggregated 参数，还有额外的输入。

这可能吗？如果可以，我在哪里可以找到文档？我的 Google-fu 让我失望了。

额外的皱纹

我提供的 PL/SQL 过程（实际上）是不确定的，它的输出具有必须保留的统计属性。例如，如果A={A[0], A[1], A[3]} 是一个特定班级的原始成绩，B=f(A) 是在 1:00 对A 调用过程的结果，而C=f(A) 是在A 调用过程的结果在 1:15 时，B={B[0],B[1],B[2]} 和 C={C[0],C[1],C[2]} 都是可以使用的输出，但是像 {C[0],B[1],C[2]} 这样的元素的混合是不可接受的。

这样做的结果是必须在每个分区上只调用一次该过程。 （好吧，从技术上讲，它可以随意调用多次，但分区的所有结果都必须来自同一次调用）。

假设，例如，我提供的程序操作如下：它接受一个等级集合作为IN OUT 参数，然后将随机选择的其中一个等级设置为 100。所有其他等级设置为零。在下午 1 点运行它可能会导致 Alice 有唯一的及格分数，而在下午 1:01 运行它可能会导致 Bob 有唯一的及格分数。无论如何，应该是每个班级只有一个学生通过，不多也不少。

Answer 1

Data Cartridge Developer's Guide 涵盖了这些主题。 This section 讨论用户定义的分析函数。

Answer 2

我发现创建具有多个参数的自定义聚合的唯一方法是创建一个具有所需元素数量的新 TYPE，然后将该类型的实例传递给聚合：

首先定义结构来保存你需要的所有“参数”：

create or replace type wrapper_type as object
(
   raw_grade integer,
   config_data varchar
);
/

然后创建你的聚合：

CREATE OR REPLACE TYPE analytic_wrapper AS OBJECT
(
  .. variables you might need

  STATIC FUNCTION ODCIAggregateInitialize(actx IN OUT wrapper_type) RETURN NUMBER,
  MEMBER FUNCTION ODCIAggregateIterate(self  IN OUT wrapper_type, val IN wrapper_type) RETURN NUMBER,
  MEMBER FUNCTION ODCIAggregateTerminate(self IN wrapper_type, returnValue OUT number, flags IN NUMBER) RETURN NUMBER,
  MEMBER FUNCTION ODCIAggregateMerge(self  IN OUT wrapper_type, ctx2 IN  wrapper_type) RETURN NUMBER
);
/

那么你需要在type body中实现实际的聚合逻辑。完成后，您可以使用以下内容：

select analytic_wrapper(wrapper_type(G.raw_grade, P.course_config_data))
from ... 

以上内容或多或少是凭记忆写的，所以我很确定它充满了语法错误，但它应该可以帮助你入门。

更多细节和示例在手册中：http://docs.oracle.com/cd/E11882_01/appdev.112/e10765/aggr_functions.htm#ADDCI026

The manual 声明这样的聚合可以用作分析函数：

当使用自定义聚合作为解析函数时，会针对每一行对应的窗口计算聚合

Answer 3

我也有同样的需要。我正在发布一种似乎可行的方法（它与我迄今为止尝试过的所有案例的本机 Oracle ratio_to_report 函数一致）。

我担心的是它依赖于 ODCIIterate 和 ODCITerminate 方法总是以相同的顺序调用的“事实”。我没有任何理由相信情况总是如此。我可能会记录一个 SR，因为如果没有 Oracle 的说明，我认为我不能使用这个版本。

不过，我还是发布了代码，因为它确实代表了问题的答案。

Caveat #1 - 此代码将状态存储在 PL/SQL 包中。我讨厌这个，但我没有其他选择，因为 ODCITerminate 仅将 SELF 传递为 IN，而不是 IN OUT。除了丑陋之外，这意味着您不能在同一个查询中多次使用自定义分析函数（因为它们的状态将混合在一起）。我确信可以围绕这个限制进行编码（例如，给每个 ODCI 上下文一个唯一的值，并为每个唯一的上下文保留单独的状态）。

Caveat #2 -- 我的测试用例使用了 PARALLEL 查询。我可以从explain plan 看到它正在并行运行。但是，它似乎并没有实例化和合并多个上下文，我真的很想测试它，因为如果有任何东西破坏了这种方法，那就是这样。

这是代码。

CREATE OR REPLACE TYPE matt_ratio_to_report_rec AS OBJECT
  ( key     VARCHAR2(80),
    value   NUMBER );

CREATE OR REPLACE PACKAGE matt_ratio_to_report_state AS
  TYPE values_tab_t IS TABLE OF matt_ratio_to_report_rec INDEX BY BINARY_INTEGER;
  TYPE index_tab_t IS TABLE OF NUMBER INDEX BY VARCHAR2(80);
  G_VALUES_TAB values_tab_t;
  G_INDEX_TAB index_tab_t;
  G_ITERATOR_POSITION NUMBER;
  G_TERMINATOR_POSITION NUMBER;
END matt_ratio_to_report_state;
/


CREATE OR REPLACE TYPE matt_ratio_to_report_impl AS OBJECT
(
  window_sum      NUMBER,
  CONSTRUCTOR FUNCTION matt_ratio_to_report_impl(SELF IN OUT NOCOPY matt_ratio_to_report_impl ) RETURN SELF AS RESULT,  
-- Called to initialize a new aggregation context
-- For analytic functions, the aggregation context of the *previous* window is passed in, so we only need to adjust as needed instead 
-- of creating the new aggregation context from scratch
  STATIC FUNCTION ODCIAggregateInitialize (sctx IN OUT matt_ratio_to_report_impl) RETURN NUMBER,
-- Called when a new data point is added to an aggregation context  
  MEMBER FUNCTION ODCIAggregateIterate (self IN OUT matt_ratio_to_report_impl, value IN matt_ratio_to_report_rec ) RETURN NUMBER,
-- Called to return the computed aggragate from an aggregation context
  MEMBER FUNCTION ODCIAggregateTerminate (self IN matt_ratio_to_report_impl, returnValue OUT NUMBER, flags IN NUMBER) RETURN NUMBER,
-- Called to merge to two aggregation contexts into one (e.g., merging results of parallel slaves) 
  MEMBER FUNCTION ODCIAggregateMerge (self IN OUT matt_ratio_to_report_impl, ctx2 IN matt_ratio_to_report_impl) RETURN NUMBER,
  -- ODCIAggregateDelete
  MEMBER FUNCTION ODCIAggregateDelete(self IN OUT matt_ratio_to_report_impl, value matt_ratio_to_report_rec) RETURN NUMBER  
);

/

CREATE OR REPLACE TYPE BODY matt_ratio_to_report_impl IS

CONSTRUCTOR FUNCTION matt_ratio_to_report_impl(SELF IN OUT NOCOPY matt_ratio_to_report_impl ) RETURN SELF AS RESULT IS
BEGIN
  SELF.window_sum := 0;
  matt_ratio_to_report_state.G_VALUES_TAB.DELETE;
  matt_ratio_to_report_state.G_INDEX_TAB.DELETE;
  matt_ratio_to_report_state.G_ITERATOR_POSITION := 0;
  matt_ratio_to_report_state.G_TERMINATOR_POSITION := 0;  
  RETURN;
END;


STATIC FUNCTION ODCIAggregateInitialize (sctx IN OUT matt_ratio_to_report_impl) RETURN NUMBER IS
BEGIN
  DBMS_OUTPUT.PUT_LINE('ODCIAggregateInitialize(' || sctx.window_sum);
  sctx := matt_ratio_to_report_impl ();
  RETURN ODCIConst.Success;
END;


MEMBER FUNCTION ODCIAggregateIterate (self IN OUT matt_ratio_to_report_impl, value IN matt_ratio_to_report_rec ) RETURN NUMBER IS
BEGIN
   DBMS_OUTPUT.PUT_LINE('ODCIAggregateIterate(' || self.window_sum || ' - ' || value.key || ', ' || value.value || ')');

  -- Increment sum
  self.window_sum := self.window_sum + value.value;

  matt_ratio_to_report_state.G_ITERATOR_POSITION := matt_ratio_to_report_state.G_ITERATOR_POSITION + 1;
  matt_ratio_to_report_state.G_VALUES_TAB(matt_ratio_to_report_state.G_ITERATOR_POSITION) := value;
  matt_ratio_to_report_state.G_INDEX_TAB(value.key) := matt_ratio_to_report_state.G_ITERATOR_POSITION;

  RETURN ODCIConst.Success;
END;

MEMBER FUNCTION ODCIAggregateTerminate (self IN matt_ratio_to_report_impl, returnValue OUT NUMBER, flags IN NUMBER) RETURN NUMBER IS
BEGIN
   DBMS_OUTPUT.PUT_LINE('ODCIAggregateTerminate(' || self.window_sum || ' - flags: ' || flags || ')');
  IF flags = 1 THEN
    matt_ratio_to_report_state.G_TERMINATOR_POSITION := matt_ratio_to_report_state.G_TERMINATOR_POSITION + 1;
    returnValue := matt_ratio_to_report_state.G_VALUES_TAB( matt_ratio_to_report_state.G_TERMINATOR_POSITION).value / self.window_sum; --self.x_list;
  END IF;
  RETURN ODCIConst.Success;
EXCEPTION
  WHEN others THEN 
    DBMS_OUTPUT.PUT_LINE(DBMS_UTILITY.FORMAT_ERROR_STACK || ' ' || DBMS_UTILITY.FORMAT_ERROR_BACKTRACE);
    RETURN ODCIConst.Success;
END;

MEMBER FUNCTION ODCIAggregateMerge (self IN OUT matt_ratio_to_report_impl, ctx2 IN matt_ratio_to_report_impl) RETURN NUMBER IS
BEGIN
   DBMS_OUTPUT.PUT_LINE('ODCIAggregateMerge(' || self.window_sum || ' - ' || ctx2.window_sum || ')');
  -- Increment sums
  self.window_sum := self.window_sum + ctx2.window_sum;

  RETURN ODCIConst.Success;
END;

-- ODCIAggregateDelete
MEMBER FUNCTION ODCIAggregateDelete(self IN OUT matt_ratio_to_report_impl, value matt_ratio_to_report_rec) RETURN NUMBER IS
BEGIN
   DBMS_OUTPUT.PUT_LINE('ODCIAggregateDelete(' || self.window_sum || ' - ' || value.key || ', ' || value.value || ')');
  -- Decrement sums
  matt_ratio_to_report_state.G_VALUES_TAB.DELETE(matt_ratio_to_report_state.G_INDEX_TAB(value.key));
  matt_ratio_to_report_state.G_INDEX_TAB.DELETE(value.key);

  self.window_sum := self.window_sum - value.value;
END;  

END;
/

CREATE OR REPLACE FUNCTION matt_ratio_to_report ( input matt_ratio_to_report_rec) RETURN NUMBER
PARALLEL_ENABLE AGGREGATE USING matt_ratio_to_report_impl;
/


CREATE TABLE matt_test_data ( x ) PARALLEL 4
AS SELECT rownum FROM DUAL CONNECT BY ROWNUM <= 50000;

with test as (
select d.x, sum(d.x) over ( partition by mod(d.x,5) order by d.x desc ) running_sum,
ratio_to_report(d.x) over ( partition by mod(d.x,500) ) oracle_rr,
matt_ratio_to_report( matt_ratio_to_report_rec(to_char(d.x), d.x) ) over ( partition by mod(d.x,500) ) custom_rr
--matt_ratio_to_report( matt_ratio_to_report_rec(to_char(d.x), d.x) ) over ( partition by mod(d.x,500) ORDER BY d.x ASC ROWS BETWEEN UNBOUNDED PRECEDING AND UNBOUNDED FOLLOWING) custom_rr_windowed 
from matt_test_data d )
SELECT /*+ PARALLEL */ test.*, case when test.oracle_rr != test.custom_rr then 'Mismatch!' Else null END test_results from test 
--where oracle_rr != custom_rr
ORDER BY test_results nulls last, x
;

Answer 4

这个版本没有我之前回答的警告，虽然它会更慢并且更难使用。 ODCIAggregateDelete 中循环的大部分速度都很慢 - 您可能可以在那里找到不需要循环遍历整个集合的改进。

无论如何，这个版本创建了一个自定义分析函数，模仿了 Oracle 的原生 COLLECT 函数。因此，与其尝试创建一个自定义分析函数来计算我们想要的实际值，它只是计算窗口中的行集。

然后，对于每一行，我们将行数据和自定义“COLLECT”分析的结果传递给计算我们想要的值的常规函数​​。

这是代码。 （注意：您最初的问题还询问了多个参数。很简单——只需将您想要的所有字段放入matt_ratio_to_report_rec。）（另外，对对象名称感到抱歉——我在其他开发人员知道的所有内容上加上我的名字询问对象是否导致他们出现问题。）

-- This is the input data to the analytic function
--DROP TYPE matt_ratio_to_report_rec;
CREATE OR REPLACE TYPE matt_ratio_to_report_rec AS OBJECT
  ( value   NUMBER );

-- This is a collection of input data  
--DROP TYPE matt_ratio_to_report_tab;
CREATE OR REPLACE TYPE matt_ratio_to_report_tab AS TABLE OF matt_ratio_to_report_rec;


-- This object type implements a custom analytic that acts as an analytic version of Oracle's COLLECT function
--DROP TYPE matt_ratio_to_report_col_impl;
CREATE OR REPLACE TYPE matt_ratio_to_report_col_impl AS OBJECT (
  analytics_window    matt_ratio_to_report_tab,
  CONSTRUCTOR FUNCTION matt_ratio_to_report_col_impl(SELF IN OUT NOCOPY matt_ratio_to_report_col_impl ) RETURN SELF AS RESULT,  
-- Called to initialize a new aggregation context
-- For analytic functions, the aggregation context of the *previous* window is passed in, so we only need to adjust as needed instead 
-- of creating the new aggregation context from scratch
  STATIC FUNCTION ODCIAggregateInitialize (sctx IN OUT matt_ratio_to_report_col_impl) RETURN NUMBER,
-- Called when a new data point is added to an aggregation context  
  MEMBER FUNCTION ODCIAggregateIterate (self IN OUT matt_ratio_to_report_col_impl, value IN matt_ratio_to_report_rec ) RETURN NUMBER,
-- Called to return the computed aggragate from an aggregation context
  MEMBER FUNCTION ODCIAggregateTerminate (self IN matt_ratio_to_report_col_impl, returnValue OUT matt_ratio_to_report_tab, flags IN NUMBER) RETURN NUMBER,
-- Called to merge to two aggregation contexts into one (e.g., merging results of parallel slaves) 
  MEMBER FUNCTION ODCIAggregateMerge (self IN OUT matt_ratio_to_report_col_impl, ctx2 IN matt_ratio_to_report_col_impl) RETURN NUMBER,
  -- ODCIAggregateDelete
  MEMBER FUNCTION ODCIAggregateDelete(self IN OUT matt_ratio_to_report_col_impl, value matt_ratio_to_report_rec) RETURN NUMBER  
);

CREATE OR REPLACE TYPE BODY matt_ratio_to_report_col_impl IS

CONSTRUCTOR FUNCTION matt_ratio_to_report_col_impl(SELF IN OUT NOCOPY matt_ratio_to_report_col_impl ) RETURN SELF AS RESULT IS
BEGIN
  SELF.analytics_window := new matt_ratio_to_report_tab();
  RETURN;
END;


STATIC FUNCTION ODCIAggregateInitialize (sctx IN OUT matt_ratio_to_report_col_impl) RETURN NUMBER IS
BEGIN
  DBMS_OUTPUT.PUT_LINE('ODCIAggregateInitialize()');
  sctx := matt_ratio_to_report_col_impl ();
  RETURN ODCIConst.Success;
END;


MEMBER FUNCTION ODCIAggregateIterate (self IN OUT matt_ratio_to_report_col_impl, value IN matt_ratio_to_report_rec ) RETURN NUMBER IS
BEGIN
   DBMS_OUTPUT.PUT_LINE('ODCIAggregateIterate(' || self.analytics_window.COUNT || ')');

  -- Add record to collection
  self.analytics_window.extend();
  self.analytics_window(self.analytics_window.COUNT) :=  value;
  RETURN ODCIConst.Success;
END;

MEMBER FUNCTION ODCIAggregateTerminate (self IN matt_ratio_to_report_col_impl, returnValue OUT matt_ratio_to_report_tab, flags IN NUMBER) RETURN NUMBER IS
BEGIN
   DBMS_OUTPUT.PUT_LINE('ODCIAggregateTerminate(' || self.analytics_window.COUNT || ' - flags: ' || flags || ')');
  IF flags = 1 THEN
    returnValue := self.analytics_window;
  END IF;
  RETURN ODCIConst.Success;
EXCEPTION
  WHEN others THEN 
    DBMS_OUTPUT.PUT_LINE(DBMS_UTILITY.FORMAT_ERROR_STACK || ' ' || DBMS_UTILITY.FORMAT_ERROR_BACKTRACE);
    RETURN ODCIConst.Success;
END;

MEMBER FUNCTION ODCIAggregateMerge (self IN OUT matt_ratio_to_report_col_impl, ctx2 IN matt_ratio_to_report_col_impl) RETURN NUMBER IS
BEGIN
--   DBMS_OUTPUT.PUT_LINE('ODCIAggregateMerge(' || self.window_sum || ' - ' || ctx2.window_sum || ')');
  -- TODO: Add all elements from ctx2 window to self window
  RETURN ODCIConst.Success;
END;

-- ODCIAggregateDelete
MEMBER FUNCTION ODCIAggregateDelete(self IN OUT matt_ratio_to_report_col_impl, value matt_ratio_to_report_rec) RETURN NUMBER IS
  l_ctr NUMBER;
BEGIN
   DBMS_OUTPUT.PUT_LINE('ODCIAggregateDelete(' || self.analytics_window.COUNT || ' - ' || value.value || ')');
   l_ctr := self.analytics_window.FIRST;
   <<window_loop>>
   WHILE l_ctr IS NOT NULL LOOP
     IF ( self.analytics_window(l_ctr).value = value.value ) THEN
       self.analytics_window.DELETE(l_ctr);
       DBMS_OUTPUT.PUT_LINE('... deleted slot ' || l_ctr);
       EXIT window_loop;
     END IF;
     l_ctr := self.analytics_window.NEXT(l_ctr);
   END LOOP;
  RETURN ODCIConst.Success;

END;  

END;
/

-- This function is the analytic version of Oracle's COLLECT function
--DROP FUNCTION matt_ratio_to_report;
CREATE OR REPLACE FUNCTION matt_ratio_to_report_col ( input matt_ratio_to_report_rec) RETURN matt_ratio_to_report_tab
PARALLEL_ENABLE AGGREGATE USING matt_ratio_to_report_col_impl;
/


-- This the actual function we want
CREATE OR REPLACE FUNCTION matt_ratio_to_report ( p_row_value NUMBER, p_report_window matt_ratio_to_report_tab ) RETURN NUMBER IS
  l_report_window_sum NUMBER := 0;
  l_counter NUMBER := NULL;
BEGIN
  IF p_row_value IS NULL or p_report_window IS NULL THEN
    RETURN NULL;
  END IF;

  -- Compute window sum
  l_counter := p_report_window.FIRST;
  WHILE l_counter IS NOT NULL LOOP
    l_report_window_sum := l_report_window_sum + NVL(p_report_window(l_counter).value,0);
    l_counter := p_report_window.NEXT(l_counter);
  END LOOP;

  RETURN p_row_value / NULLIF(l_report_window_sum,0);
END matt_ratio_to_report;  



-- Create some test data
--DROP TABLE matt_test_data;
CREATE TABLE matt_test_data ( x, group# ) PARALLEL 4
AS SELECT rownum, ceil(rownum / 10) group# FROM DUAL CONNECT BY ROWNUM <= 50000;


-- TESTER 9/30
with test as (
SELECT d.x,
       CEIL (d.x / 10) group#,
       ratio_to_report (d.x) OVER (PARTITION BY d.group#) oracle_rr,
       matt_ratio_to_report (
         d.x,
         matt_ratio_to_report_col (matt_ratio_to_report_rec (d.x)) OVER (PARTITION BY d.group#)) custom_rr
FROM   matt_test_data d )
SELECT /*+ PARALLEL */ test.*, case when test.oracle_rr != test.custom_rr then 'Mismatch!' Else null END test_results from test 
--where oracle_rr != custom_rr
ORDER BY test_results nulls last, x;