PostgreSQL Oracle 兼容性之 - 自定义并行聚合函数 PARALLEL_ENABLE AGGREGATE

2 minute read

背景

Oracle，PostgreSQL均支持自定义聚合函数，例如PostgreSQL自定义聚合函数的语法如下：

Command:     CREATE AGGREGATE  
Description: define a new aggregate function  
Syntax:  
CREATE AGGREGATE name ( [ argmode ] [ argname ] arg_data_type [ , ... ] ) (  
    SFUNC = sfunc,  
    STYPE = state_data_type  
    [ , SSPACE = state_data_size ]  
    [ , FINALFUNC = ffunc ]  
    [ , FINALFUNC_EXTRA ]  
    [ , COMBINEFUNC = combinefunc ]  
    [ , SERIALFUNC = serialfunc ]  
    [ , DESERIALFUNC = deserialfunc ]  
    [ , INITCOND = initial_condition ]  
    [ , MSFUNC = msfunc ]  
    [ , MINVFUNC = minvfunc ]  
    [ , MSTYPE = mstate_data_type ]  
    [ , MSSPACE = mstate_data_size ]  
    [ , MFINALFUNC = mffunc ]  
    [ , MFINALFUNC_EXTRA ]  
    [ , MINITCOND = minitial_condition ]  
    [ , SORTOP = sort_operator ]  
    [ , PARALLEL = { SAFE | RESTRICTED | UNSAFE } ]  
)  
  
CREATE AGGREGATE name ( [ [ argmode ] [ argname ] arg_data_type [ , ... ] ]  
                        ORDER BY [ argmode ] [ argname ] arg_data_type [ , ... ] ) (  
    SFUNC = sfunc,  
    STYPE = state_data_type  
    [ , SSPACE = state_data_size ]  
    [ , FINALFUNC = ffunc ]  
    [ , FINALFUNC_EXTRA ]  
    [ , INITCOND = initial_condition ]  
    [ , PARALLEL = { SAFE | RESTRICTED | UNSAFE } ]  
    [ , HYPOTHETICAL ]  
)  
  
or the old syntax  
  
CREATE AGGREGATE name (  
    BASETYPE = base_type,  
    SFUNC = sfunc,  
    STYPE = state_data_type  
    [ , SSPACE = state_data_size ]  
    [ , FINALFUNC = ffunc ]  
    [ , FINALFUNC_EXTRA ]  
    [ , COMBINEFUNC = combinefunc ]  
    [ , SERIALFUNC = serialfunc ]  
    [ , DESERIALFUNC = deserialfunc ]  
    [ , INITCOND = initial_condition ]  
    [ , MSFUNC = msfunc ]  
    [ , MINVFUNC = minvfunc ]  
    [ , MSTYPE = mstate_data_type ]  
    [ , MSSPACE = mstate_data_size ]  
    [ , MFINALFUNC = mffunc ]  
    [ , MFINALFUNC_EXTRA ]  
    [ , MINITCOND = minitial_condition ]  
    [ , SORTOP = sort_operator ]  
)  

Oracle的语法与之有所不同，并且支持并行。

PostgreSQL 9.6开始，也支持了并行，如上语法中包含了并行的语法。

例子

Oracle 的一个自定义聚合函数如下：

CREATE OR REPLACE FUNCTION launch_concat (  
    value Varchar2  -- 关键点  
)  
RETURN Varchar2 AUTHID definer  
PARALLEL_ENABLE  -- 并行  
AGGREGATE        -- 并行  
USING string_sum_obj;  -- 关键点  

在PostgreSQL中可以创建一个类似的聚合函数（我们先不管是否与Oracle该例子中的string_sum_obj实现是否一致），语法如下：

create aggregate launch_concat(text,text) (  
  sfunc = pg_catalog.string_agg_transfn,  
  stype = internal,  
  FINALFUNC = pg_catalog.string_agg_finalfn  
);  

例子

postgres=# select launch_concat(id::text, ',') from generate_series(1,10) t(id);  
    launch_concat       
----------------------  
 1,2,3,4,5,6,7,8,9,10  
(1 row)  

如果要达到与Oracle聚合一样的效果，需要解析Oracle的这个基础函数string_sum_obj的内容，到底要实现什么效果，根据需求在PostgreSQL中实现一遍即可，然后代替掉：

  sfunc = pg_catalog.string_agg_transfn,  
  stype = internal,  
  FINALFUNC = pg_catalog.string_agg_finalfn  

并行如何支持？

1、扫描并行

create aggregate launch_concat(text,text) (  
  sfunc = pg_catalog.string_agg_transfn,  
  stype = internal,  
  FINALFUNC = pg_catalog.string_agg_finalfn,   
  PARALLEL=safe  
);  

2、聚合并行

得实现COMBINEFUNC，即将多个WORKER聚合的结果合并的函数。

具体参考

《PostgreSQL 10 自定义并行计算聚合函数的原理与实践 - (含array_agg合并多个数组为单个一元数组的例子)》