PostgreSQL知识分享-第34期

本期分享如下：

• date_trunc示例
• upsert示例两则
• archive_command预防core dump file
• 模拟MySQL的substring_index函数
• 模拟MySQL的group_concat函数

一、date_trunc示例

问题摘要来源：Bruce Momjian Indexing timestamps

时区类型(TIMESTAMP WITH TIME ZONE)在PG里是比较常见的一种类型，它存的值精确到时分秒，同时也包含时区信息。

不过有些场景查询并不需要特别精确，例如只需查询到天的粒度。接下来看下面这个示例：

CREATE TEMPORARY TABLE date_test (event_time TIMESTAMP WITH TIME ZONE);
 
INSERT INTO date_test
SELECT
(
    SELECT '2023-03-01 00:00:00'::timestamptz +
           (floor(random() *
            (extract(EPOCH FROM '2023-04-01'::timestamptz) -
             extract(EPOCH FROM '2023-03-01'::timestamptz)) +
            b * 0)::integer || 'seconds')::interval
)
FROM generate_series(1, 100000) AS a(b);
 
ANALYZE date_test;
复制

对date_test表查询的时间粒度只关注到天(date)，首先会考虑如下的索引创建方式：

CREATE INDEX i_date_test ON date_test ((event_time::date));
复制

不过我们会得到如下的错误信息：

ERROR:  functions in index expression must be marked IMMUTABLE
复制

索引创建失败的原因是：不同的会话可能使用不同的时区值，这会导致不一致的结果。

一种解决方式是显式设置时区值：

SHOW timezone;
     TimeZone     
------------------
 America/New_York
 
CREATE INDEX i_date_test_nyc ON date_test (((event_time AT TIME ZONE 'America/New_York')::date));
复制

这种方式只适用于固定的时区值匹配。

类似的方式可以使用16版本提供的date_trunc函数：

CREATE INDEX i_date_test_trunc ON date_test ((date_trunc('day', event_time, 'America/New_York')));
复制

date_trunc函数除了支持设置时区，对时间的精度选择也更灵活，可选择day、month等。

二、upsert语法示例

在PG里upsert相当于insert + update的组合，当数据行insert发生主键或唯一约束重复时，可触发update操作，参考语法如下：

insert into ... ON CONFLICT(XXX) do update set columnN=excluded.columnN;
复制

示例一：

create table test_upsert (id int primary key,code varchar unique,info varchar);

insert into test_upsert (id,code,info) 
    values (1,'pg','info');
复制

上面插入一条主键id为1的数据后，再次插入id为1的数据时可使用upsert语句正常执行

insert into test_upsert (id,code,info) 
    values (1,'pg','info extra1') 
    on conflict(id) 
    do update set info=excluded.info,id=excluded.id;

insert into test_upsert (id,code,info) 
    values (1,'pg','info extra2') 
    on conflict(id) 
    do update set info=excluded.info,code=excluded.code;
复制

upsert语句允许在update子句里更新主键id字段亦或唯一约束code字段。

示例二：

create table test_upsert2 (
    id int primary key,
    code1 varchar not null,
    code2 varchar not null,
    info varchar,
    unique(code1,code2)
);

insert into test_upsert2 (id,code1,code2,info) values (1,'MySQL','PG','info');
复制

test_upsert2表里使用了多字段唯一索引，下面upsert语句可以指定冲突的条件是unique(code1,code2)，而非主键id，同时在update子句也可对主键id或组合唯一索引的部分字段code1进行操作

insert into test_upsert2 (id,code1,code2,info) 
    values (1,'openGauss','PG','pg db extra1') 
    on conflict(code1,code2) 
    do update set info=excluded.info, 
                  id=excluded.id,
                  code1=excluded.code1;
复制

通过上面两个示例可以看出upsert语法考虑比较全面，允许显式指定主键冲突或唯一约束，而且也可对主键及唯一约束字段进行更新。

三、archive_command预防core dump file

问题摘要来源：夜雨成诗 PgSQL · 答疑解惑 · 归档进程cp命令的core文件追查

core文件的产生原因可以总结为，发生OOM Kill时，PG主进程会向所有子进程和子进程所拥有的Process Group发送Kill -3信号；另一方面，归档进程会fork子进程来执行归档命令，此子进程在归档进程的Process Group里面，故也收到了Kill -3信号。而且该进程会对信号执行缺省动作即产生core文件。

如果我们设置core file的size limit为0，就会阻止core文件产生。而对于出问题的PG实例，我们是在pg_ctl启动进程时加入了-c选项，将core file 的size limit去除；而所有Postmaster的子进程和孙子进程，又继承了父进程的size limit，导致core file产生。所以，此问题的一个规避方法为，对archive_command做如下设置：

archive_command='ulimit -c 0 && cp %p /u01/tmp/%f'
复制

这样在cp命令被归档进程调用时，其core file的size limit为0，即便收到SIGQUIT信号，也不会打印core dump file。

四、模拟MySQL的substring_index函数

MySQL的substring_index函数使用示例如下：

MariaDB [test]> SELECT substring_index('www.mysql.com', '.', 2);
+------------------------------------------+
| substring_index('www.mysql.com', '.', 2) |
+------------------------------------------+
| www.mysql                                |
+------------------------------------------+
1 row in set (0.000 sec)

MariaDB [test]> SELECT substring_index('www.mysql.com', '.', -2);
+-------------------------------------------+
| substring_index('www.mysql.com', '.', -2) |
+-------------------------------------------+
| mysql.com                                 |
+-------------------------------------------+
1 row in set (0.000 sec)
复制

由于MySQL的低版本没有arm架构，我的实验环境用MariaDB模拟。

在PG里有如下几种方式：

场景一：正向只获取一个元素，使用split_part函数实现

postgres=# SELECT split_part('www.mysql.com', '.', 1);
 split_part 
------------
 www
(1 row)
复制

场景二：反向只获取一个元素，使用split_part + reverse函数实现

postgres=# SELECT reverse(split_part(reverse('www.mysql.com'), '.', 1));
 reverse 
---------
 com
(1 row)
复制

场景三：正向获取多个元素，使用regexp_split_to_array + array_to_string函数实现

postgres=# SELECT array_to_string((regexp_split_to_array('www.mysql.com', '\.'))[:2],'.');
 array_to_string 
-----------------
 www.mysql
(1 row)
复制

注意：
1.regexp_split_to_array函数里使用的分隔符“点号”为特殊符号，需要加转义符号。
2.数组元素上下边界引用使用了简写方式，只指定了下边界，版本9.6引入的特性。

https://www.postgresql.org/docs/9.6/release-9-6.html

E.25.3.6. Data Types

Allow omitting one or both boundaries in an array slice specifier, e.g., array_col[3:] (Yury Zhuravlev)

Omitted boundaries are taken as the upper or lower limit of the corresponding array subscript. This allows simpler specification for many common use-cases.

场景四：正反向获取多个元素，使用自定义函数：

CREATE OR REPLACE FUNCTION public.substring_index (
  str text,
  delim text,
  count integer = 1,
  out substring_index text
)
RETURNS text AS
$body$
BEGIN
  IF count > 0 THEN
    substring_index = array_to_string((string_to_array(str, delim))[:count], delim);
  ELSE
    DECLARE
      _array TEXT[];
    BEGIN
      _array = string_to_array(str, delim);
      substring_index = array_to_string(_array[array_length(_array, 1) + count + 1:], delim);    
    END;  
  END IF;
END;
$body$
LANGUAGE 'plpgsql'
IMMUTABLE
CALLED ON NULL INPUT
SECURITY INVOKER
COST 5;
复制

函数定义来源于：

https://stackoverflow.com/questions/19230584/emulating-mysqls-substring-index-in-pgsql

五、模拟MySQL的group_concat函数

MySQL的group_concat函数，在PG里可以使用string_agg替代，示例如下：

postgres=# create table foo(id int,info varchar);
CREATE TABLE
postgres=# insert into foo values(100,'1a');
INSERT 0 1
postgres=#  insert into foo values(100,'2b');
INSERT 0 1
postgres=# insert into foo values(100,'3c');
INSERT 0 1
postgres=# insert into foo values(200,'2a');
INSERT 0 1
postgres=# insert into foo values(200,'3b');
INSERT 0 1
postgres=#  insert into foo values(200,'4c');
INSERT 0 1
postgres=#  select id,string_agg(info,',') from foo group by id order by id;
 id  | string_agg 
-----+------------
 100 | 1a,2b,3c
 200 | 2a,3b,4c
(2 rows)
复制

那我们参考string_agg函数的定义，再创建一个别名函数group_concat即可。

使用元命令\sf查看其定义，发现函数有重载

postgres=# \sf string_agg
ERROR:  more than one function named "string_agg"
复制

string_agg的输入参数有二进制和文本类型两种

postgres=# \df string_agg
                            List of functions
   Schema   |    Name    | Result data type | Argument data types | Type 
------------+------------+------------------+---------------------+------
 pg_catalog | string_agg | bytea            | bytea, bytea        | agg
 pg_catalog | string_agg | text             | text, text          | agg
(2 rows)
复制

带上文本类型的输入参数，再次查看定义：

postgres=# \sf string_agg(text,text)
ERROR:  "string_agg" is an aggregate function
复制

元命令\sf只能查看普通函数的定义，不能查看聚合函数的定义

那我们根据pg_aggregate系统表记录string_agg的信息，凭接还原其定义：

SELECT format('CREATE AGGREGATE %s (SFUNC = %s ,STYPE = %s , FINALFUNC = %s%s%s%s%s);'
            , aggfnoid::regprocedure
            , aggtransfn
            , aggtranstype::regtype
            , aggfinalfn
            , ', COMBINEFUNC = '  || aggcombinefn
            , ', SERIALFUNC = '  || aggserialfn
            , ', DESERIALFUNC = '  || aggdeserialfn
            , ', PARALLEL = SAFE '
              ) AS ddl_agg
FROM   pg_aggregate
WHERE  aggfnoid = 'string_agg(text,text)'::regprocedure;
复制

在PG16里执行得到如下结果：

CREATE AGGREGATE string_agg(text,text) (
    SFUNC = string_agg_transfn ,
    STYPE = internal , 
    FINALFUNC = string_agg_finalfn , 
    COMBINEFUNC = string_agg_combine , 
    SERIALFUNC = string_agg_serialize, 
    DESERIALFUNC = string_agg_deserialize,
    PARALLEL = SAFE 
);
复制

修改上面语句里聚合函数的名称，改为public.group_concat后可以创建成功

CREATE AGGREGATE public.group_concat(text,text) (
    SFUNC = string_agg_transfn ,
    STYPE = internal , 
    FINALFUNC = string_agg_finalfn , 
    COMBINEFUNC = string_agg_combine , 
    SERIALFUNC = string_agg_serialize, 
    DESERIALFUNC = string_agg_deserialize,
    PARALLEL = SAFE 
);
复制

在较低的版本，例如PG 9.2，使用如下语句创建

CREATE AGGREGATE public.group_concat(text,text) (
    SFUNC = string_agg_transfn ,
    STYPE = internal , 
    FINALFUNC = string_agg_finalfn
);
复制

最后验证group_concat，执行结果如下：

postgres=# select id,group_concat(info,',') from foo group by id order by id;
 id  | group_concat 
-----+-----------
 100 | 1a,2b,3c
 200 | 2a,3b,4c
(2 rows)
复制

关联推荐

• PostgreSQL知识问答分享-第35期
• PostgreSQL知识问答分享-第34期
• PostgreSQL知识问答分享-第33期
• PostgreSQL知识问答分享-第32期
• PostgreSQL知识问答分享-第31期
• PostgreSQL知识问答分享-第30期
• PostgreSQL知识问答分享-第29期
• PostgreSQL知识问答分享-第28期
• PostgreSQL知识问答分享-第27期
• PostgreSQL知识问答分享-第26期

如果有任何问题需要讨论交流的朋友，欢迎添加本人微信号skypkmoon。