使用hstore实现postgresql自定义聚合函数

最近工作中遇到一个变态的报表查询需求，为了简化业务需求，说明如下：

 

CREATE TABLE public.book (
  bookid INTEGER NOT NULL,
  bookname CHARACTER VARYING(255) NOT NULL,
  authors CHARACTER VARYING(255) NOT NULL,
  info CHARACTER VARYING(255) NOT NULL,
  comment CHARACTER VARYING(255) NOT NULL,
  year_publication DATE NOT NULL,
  publisher CHARACTER VARYING(10) -- 出版社
);
COMMENT ON COLUMN public.book.publisher IS '出版社';

 有一张表，出版社书籍出版表，bookname:书名，authors:作者，year_publication:出版日期，publisher:出版社名称，info和comment字段可以忽略不计，数据如下：

 

 

INSERT INTO public.book (bookid, bookname, authors, info, comment, year_publication, publisher) VALUES (5, 'c++', 'lisi', ' ', ' ', '2016-12-07', 'pub_1');
INSERT INTO public.book (bookid, bookname, authors, info, comment, year_publication, publisher) VALUES (4, 'php', 'lisi', ' ', ' ', '2016-12-08', 'pub_1');
INSERT INTO public.book (bookid, bookname, authors, info, comment, year_publication, publisher) VALUES (2, 'c', 'zhangsan', ' ', ' ', '2016-12-06', 'pub_1');
INSERT INTO public.book (bookid, bookname, authors, info, comment, year_publication, publisher) VALUES (3, 'python', 'lisi', ' ', ' ', '2016-12-09', 'pub_1');
INSERT INTO public.book (bookid, bookname, authors, info, comment, year_publication, publisher) VALUES (1, 'java', 'zhangsan', ' ', ' ', '2016-12-07', 'pub_1');

 

 

 

需求简化如下：

查询每个作者最早出版的书名，要求能根据authros或者publisher字段group by

举个例子，表中的数据，如果根据authors聚合，那么结果如下：

authros	bookname
zhangsan	c
lisi	c++

如果根据publisher字段group by，那么最终返回应该是一条记录，如下

publisher	bookname
c,c++	pub_1

即按照publisher字段group by时要看有几位作者，取每个作者最早出版的书名，拼接后返回。

这个如果用sql实现的话还是比较复杂的，领导要求用聚合函数实现，然后看性能如何。数据库用的是postgresql。

postgresql中自定义聚合函数定义如下:

 

在PostgreSQL里的聚合是依据状态值 和状态转换函数定义的。 也就是说，聚合操作使用一个随着每个输
入行被处理而变化的状态值。 要定义一个新的聚合函数，就要选择表示状态值的数据类型、状态初始值、
状态转换函数。 该状态转换函数接受之前的状态值和聚合的输入值作为当前行， 并返回一个新的状态值。
 还可以声明一个最终处理函数， 用于对付期望的聚合结果不同于需要保留在状态值中数据的情况。 
该最终处理函数接受最后的状态值并返回任何想要的作为聚合的结果。 一般而言，状态和最终函数只是
普通函数，也可以用在聚合的外面 （实际上，创建特殊的只能作为聚合的一部分调用的转换函数通常
对性能有帮助）。
因此，除了被聚合用户看到的参数和结果数据类型外，还有一种内部状态值数据类型， 这种类型可能与
参数和结果类型都不一样。

 以上定义来自postgresql手册翻译，连接如下

 

http://postgres.cn/docs/9.4/xaggr.html

以下是其中一个示例：

 

CREATE AGGREGATE avg (float8)
(
    sfunc = float8_accum,
    stype = float8[],
    finalfunc = float8_avg,
    initcond = '{0,0,0}'
);

 聚合函数的原理如下：

 

1、聚合函数包括aggregate 函数(外层包装)、sfunc和finalfunc这几个函数，其中finalfunc是可选的；通过AGGREGATE关键字定义聚合函数，聚合函数内部需要通过sfunc关键字指定sfunc、finalfunc以及一个中间状态变量；

2、sfunc是状态转换函数，对于分组后的数据，每一条记录都会调用一次sfunc，处理的中间结果可以存放在stype指定的中间变量中，在进行下一次sfunc调用时，会自动将新记录的值以及中间变量传递进去；

3、finalfunc是最终处理函数，接收参数为最后的状态值，可以对这个最后状态值处理，最后状态值是这组记录经过sfunc函数处理的结果

4、如果有finalfunc，那么这个聚合函数以finalfunc函数的返回值作为结果，如果没有定义finalfunc，那么聚合函数以sfunc函数处理的stype中间状态的最后的状态返回。

 

现在我们来分析下需求：

根据publisher字段group by的时候，所有5条数据分成一组，因此会调用sfunc5次，调用finalfunc一次，在调用sfunc的时候，需要记录有那几个不同的authors，对于每个author，需要比较对应的日期，记录最早的日期；而在finalfunc调用的时候，需要对前面的数据汇总，把不同作者最早的书名拼接起来，返回。例如：在调用完sfunc函数后，数据结构中应该记录如下：

bookname	authors	year_publication
c	zhangsan	2016-12-06
c++	lisi	2016-12-07

而在记录的过程中，刚开始数据结构中的数据时空的，sfunc处理记录的顺序可能是不定的，比如第一次调用sfunc时，中间状态是空的，处理的记录是（'张三','java','2016-12-07'）,那此时需要在数据结构中记录下来，等第二次调用sfunc，传进去记录 （'张三','c','2016-12-06'）时，要将当前记录的时间同之前的比较，如果早于之前的时间，那么需要更新数据结构中的记录为当前的。

 

在sfunc全都调用完后，一个group by分组会调用一次finalfunc函数，在finalfunc函数中需要对数据结构中的数据汇总，遍历所有数据，把bookname字段拼接，最终返回。而整个聚合函数的返回结果就是这边finalfunc函数返回的结果。

 

对于存储中间状态的数据结构，之前老的做法是使用表存储，表能存储多条记录，这能实现。但是现在，领导说表存储比较慢，看看能不能不用表实现。

 

找了一圈，看到postgresql有个hstore扩展，研究了下，这是一个类似hashmap的数据结构。

首先要安装hstore，linux环境下需要安装postgresql94-contrib-9.4.10-1PGDG.rhel6.x86_64.rpm，一下几个是测试时完整的安装文件：

 

rpm -ivh postgresql94-libs-9.4.10-1PGDG.rhel6.x86_64.rpm
rpm -ivh postgresql94-9.4.10-1PGDG.rhel6.x86_64.rpm
rpm -ivh postgresql94-server-9.4.10-1PGDG.rhel6.x86_64.rpm
rpm -ivh postgresql94-contrib-9.4.10-1PGDG.rhel6.x86_64.rpm

安装时依次按照以上顺序安装。

 

安装完后启动postgresql后，需要添加hstore扩展,切换的默认的postgres账户，启动psql，执行以下命令：

create extension hstore;

安装完后，如果你的数据库是在hstore之前创建的，那么还不够，hstore还不能使用，可以使用如下命令添加hstore：

psql database -c 'create extension hstore;'

其中database是你的数据库名称，具体的可以参看这篇文章：

http://clarkdave.net/2012/09/postgresql-error-type-hstore-does-not-exist/

这样你就能在你的数据库中使用hstore了。hstore相关的用法可以参看如下文档：

https://www.postgresql.org/docs/9.0/static/hstore.html

创建的函数如下

使用hstore自定义了一个类型：

CREATE TYPE rectime_value AS (
  pubdate hstore,   --发布时间
  bookname hstore   --书名
);

 这个一个自定义类型，里面有两个hstore类型的变量，结构为author->year_publication和author->bookname，即对应的bookname、author、year_publication，两个变量key相同的为同一记录；

 

外层的聚合函数定义如下：

CREATE AGGREGATE latest_book(bookname VARCHAR, author VARCHAR, publiction DATE)
(
  SFUNC = latest_book_sfunc,
  stype = rectime_value,
  FINALFUNC = latest_book_ffunc
);

 中间状态变量stype是自定义的rectime_value类型。

 

sfunc定义如下：

CREATE OR REPLACE FUNCTION latest_book_sfunc(last rectime_value, nb VARCHAR, na VARCHAR, np DATE)
  RETURNS rectime_value
  LANGUAGE plpgsql
AS
  $function$
  DECLARE
    temp rectime_value;
    old_date DATE;
    str_bookname VARCHAR;
    str_year VARCHAR;
  BEGIN
    RAISE INFO '--ssssssssss last = %, nb = %, na = %, np = %', last, nb, na, np;
    IF last IS NULL THEN
      --RAISE INFO '--ssssssssss last = %--', last;
      str_bookname := na || '=>' || nb;
      --RAISE INFO 'str_bookname = %', str_bookname;
      str_year := na || '=>' || np;
      --RAISE INFO 'str_year = %', str_year;
      --RAISE INFO 'temp.bookname = %', temp.bookname;
      temp.bookname := (str_bookname::hstore);
      temp.pubdate := (str_year::hstore);
      RAISE INFO '********temp = %********', temp;
      RETURN temp;
    END IF;
    IF last.pubdate?na THEN
      old_date := last.pubdate -> na;
      IF np < old_date THEN
        str_bookname := na || '=>' || nb;
        str_year := na || '=>' || np;
        last.bookname := last.bookname || (str_bookname::hstore);
        last.pubdate := last.pubdate || (str_year::hstore);
      END IF;
    ELSE
      str_bookname := na || '=>' || nb;
      str_year := na || '=>' || np;
      last.bookname := last.bookname || (str_bookname::hstore);
      last.pubdate := last.pubdate || (str_year::hstore);
    END IF;
    RETURN last;
  END
  $function$;

 函数创建的语法这里不细说了，基本思路如下：

对于第一次调用sfunc，中间变量last是空的，即第一个if中，这时候直接把当前记录的值赋给temp变量，返回temp变量；在第二次调用sfunc的时候，中间变量last就是上一次调用时的temp了，即此时last中已经有值了，会执行后面的if判断；后面的if判断就是从中间变量last中判断是否是同一个作者，如果不是同一个作者，直接把当前记录添加进中间变量，如果是同一个作者，取出上次的时间，跟当前记录的时间比较，如果晚于当前时间，则用当前记录替换中间变量中的记录，这样每次调用完sfunc后，中间变量last中记录的都是每个作者最早发布的书的信息；

 

finalfunc定义如下：

CREATE OR REPLACE FUNCTION latest_book_ffunc(last rectime_value)
  RETURNS VARCHAR
  LANGUAGE plpgsql
AS
  $function$
  DECLARE
    _Cursor refcursor;
    authors VARCHAR;
    bookname VARCHAR;
    result VARCHAR DEFAULT '';
  BEGIN
    RAISE INFO '--ffffffffff-- last = %', last;
    open _Cursor for SELECT * FROM each(last.bookname);
    fetch next from _Cursor into  authors, bookname;
    while( FOUND ) loop
      --RAISE INFO 'result = %', result;
      result := result || ',' || bookname;
      fetch next from _Cursor into  authors, bookname;
    END LOOP;
    RETURN result;
  END
  $function$;

 finalfunc的作用时将中间变量last中的数据汇总，last是一个自定义类型，里面的数据时hstore类型，可以通过hstore的each函数，将结果转换成一个游标，然后循环处理。更多hstore支持的操作可以参看上面的链接。

 

至此，一个自定义的聚合函数就完成了，运行效果如下：

 

以上是本人对postgresql自定义聚合函数的理解，如有偏差，请不吝指教！