GaussDB(DWS) SQL进阶之SQL操作 - UPSERT

举报
acydy 发表于 2021/07/25 13:28:02 2021/07/25
【摘要】 【摘要】 GaussDB(DWS)支持的UPSERT功能，可以实时进行数据导入，在出现主键或者唯一索引冲突时执行更新或者忽略操作。对于实时数仓是一项非常重要的技术。

当前GaussDB(DWS)支持UPSERT功能。本片文章介绍UPSERT功能与基本用法。

1. 前言

在前一篇博文 中介绍了一种高效数据导入方法MERGE INTO。MERGE INTO中涉及两张表，并且需要关联操作。这意味着不能一条数据做一次关联，需要等待一批数据，这样才能更加高效。其中还涉及分布式JOIN。如果对数据导入实时性要求较高，那一定不要错过GaussDB(DWS)的UPSERT功能。

2. UPSERT 概念

UPSERT : UPDATE or INSERT。可以原子性的执行如下操作：插入数据时判断是否有主键或者唯一索引冲突，如果无冲突，就进行插入，否则基于已有数据进行更新。常用于数据实时入库场景。

GaussDB(DWS)有两种功能：冲突更新或者冲突忽略：

冲突更新使用场景：业务中数据可能有重复，期望在导入数据时对数据进行实时更新。比如对同一件商某些状态状态如价格、库存等持续不断进行更新。

冲突忽略使用场景：有时可能由于网络不稳定等问题，上层业务可能会出现重复的导入数据。如果在数据库已经插入数据但是业务层以为网络问题收到报错时，业务层需要额外逻辑判定数据是否已经导入。这样降低数据导入性能，并且需要业务层参与。引入冲突忽略功能后，业务层可以无感知的进行数据导入。

3. UPSERT 语法

Command:     INSERT
Description: create new rows in a table
Syntax:
[ WITH [ RECURSIVE ] with_query [, ...] ]
INSERT [ IGNORE | OVERWRITE ] INTO table_name [ AS alias ] [ ( column_name [, ...] ) ]
    { DEFAULT VALUES | VALUES {( { expression | DEFAULT } [, ...] ) }[, ...] | query }
    [ ON DUPLICATE KEY duplicate_action | ON CONFLICT [ conflict_target ] conflict_action ]
    [ RETURNING {* | {output_expression [ [ AS ] output_name ] }[, ...]} ];

where duplicate_action can be:

    UPDATE { column_name = { expression | DEFAULT } |
             ( column_name [, ...] ) = ( { expression | DEFAULT } [, ...] )
           } [, ...]

and conflict_target can be one of:

    ( { index_column_name | ( index_expression ) } [ COLLATE collation ] [ opclass ] [, ...] ) [ WHERE index_predicate ]
    ON CONSTRAINT constraint_name

and conflict_action is one of:

    DO NOTHING
    DO UPDATE SET { column_name = { expression | DEFAULT } |
                    ( column_name [, ...] ) = ( { expression | DEFAULT } [, ...] )
                  } [, ...]
              WHERE [condition]

GaussDB(DWS)支持两种风格UPSERT语法。

冲突更新 冲突忽略
不指定索引 INSERT INTO ON DUPLICATE KEY UPDATE INSERT IGNORE
可以指定列名或者约束名用于推断唯一约束 INSERT INTO ON CONFLICT(…) DO UPDATE SET INSERT INTO ON CONFLICT(…) DO NOTHING

执行UPSERT前置条件：插入的目标表必须有主键或者唯一约束。
因为使用UPSERT UPDATE功能实际是期望对重复数据进行更新。如果没有主键或者唯一约束，执行INSERT即可。

3.1 不指定索引

会在所有主键或唯一索引上检查冲突，有冲突就会忽略或者更新。
语法介绍：

  • IGNORE

    用于主键或者唯一约束冲突时忽略数据。

  • ON DUPLICATE KEY

    用于主键或者唯一约束冲突时更新冲突的数据。

    duplicate_action指定更新列和更新的数据。

3.2 指定索引

会从ON CONFLICT子句中指定列名、包含列名的表达式或者约束名上推断主键或者唯一索引。

  • ON CONFLICT

    用于主键或者唯一约束冲突时忽略或者更新冲突的数据。

    • conflict_target用于指定列名index_column_name 、包含多个列名的表达式index_expression 或者约束名字constraint_name。

    • ON CONFLICT ON CONSTRAINT 后面指定主键或者唯一约束名。

    • ON CONFLICT(…) 括号里支持列名，用于推断在该列上的主键或者唯一约束。

      作用是通过指定列名或者约束名推断主键或者唯一索引。其中index_column_name和index_expression遵循CREATE INDEX的索引列格式。列名可以是单一或者多个列名，或者由列名组成的表达式。

      所有恰好包含指定列/表达式的主键或者唯一索引（不考虑顺序）都会被推断为满足要求的索引。

    • 此种语法的冲突忽略策略可以不指定conflict_target。即 INSERT INTO ON CONFLICT DO NOTHING。

  • conflict_action

    指定主键或者唯一约束冲突时执行的策略。有两种：

    • DO NOTHING 冲突忽略。
    • DO UPDATE SET冲突更新。 后面指定更新列和更新的数据。

3.3 UPDATE子句

  • VALUES(colname)

    VALUES(…) ：引用插入的values colname列的值。里面要求是列名、复合类型的某一列。

    UPDATE c = VALUES©，含义：如果插入数据与表中已有数据有冲突，将c列更新成INSERT INTO VALUES指定的c列值。

  • EXCLUDED.colname

    表示因冲突而本该排除的数据行，与VALUES(colname) 含义相同。

    UPDATE子句中VALUES(colname)用法不支持外层嵌套函数，即不支持类似sqrt(VALUES(colname))用法。如需支持，使用EXCLUDED.colname语法。

  • WHERE子句

    用于在数据冲突时，判断是否满足指定条件。如果满足，则更新冲突数据。否则忽略。只有INSERT INTO ON CONFLICT(…) DO UPDATE SET 支持WHERE语法。

4. 实战应用

先准备好表和数据。表dst主键是a列。

gaussdb=# CREATE TABLE dst(a int primary key, b int, c int, constraint primary_dst primary key(a)) DISTRIBUTE BY HASH(a);
gaussdb=# INSERT INTO dst VALUES(1,1,1);
gaussdb=# CREATE TABLE src(a int, b int, c int) DISTRIBUTE BY HASH(a);
gaussdb=# INSERT INTO src VALUES(1,1,1),(2,2,2),(3,3,3);

4.1 不指定索引

4.1.1 冲突更新

插入两条数据，一条与表中数据有主键冲突，另一条没有。

从插入结果可以看到数据(1,1,1)被更新成(1,1,3)。无冲突数据(2,3,4)插入。下面例子含义是(1,2,3)有冲突，把冲突那行的c列值改成3。

gaussdb=# INSERT INTO dst VALUES(1,2,3),(2,3,4) ON DUPLICATE KEY UPDATE c = VALUES(c);
INSERT 0 2
gaussdb=# SELECT * FROM dst ORDER BY 1;
 a | b | c
---+---+---
 1 | 1 | 3
 2 | 3 | 4
(2 rows)

4.1.2 冲突忽略

接着执行冲突忽略语句。 (1,2,3)由于冲突没有插入，忽略了。(3,4,5)没有冲突，进行插入。

gaussdb=# INSERT IGNORE INTO dst VALUES(1,2,3),(3,4,5);
INSERT 0 1
gaussdb=# SELECT * FROM dst ORDER BY 1;
 a | b | c
---+---+---
 1 | 1 | 3
 2 | 3 | 4
 3 | 4 | 5
(3 rows)

4.2 指定索引

4.2.1 指定列名，冲突更新

一条数据冲突进行更新，一条数据无冲突进行插入。

gaussdb=# TRUNCATE dst;
gaussdb=# INSERT INTO dst VALUES(1,1,1);
gaussdb=# INSERT INTO dst VALUES(1,2,3),(2,3,4) ON CONFLICT(a) DO UPDATE SET b = EXCLUDED.b;
INSERT 0 2
gaussdb=# SELECT * FROM dst;
 a | b | c
---+---+---
 1 | 2 | 1
 2 | 3 | 4
(2 rows)

4.2.2 指定约束名

两条数据在约束primary_dst上都有冲突，更新b列。

gaussdb=# INSERT INTO dst VALUES(1,3,3),(2,4,4) ON CONFLICT ON CONSTRAINT primary_dst DO UPDATE SET b = EXCLUDED.b;
INSERT 0 2
gaussdb=# SELECT * FROM dst;
 a | b | c
---+---+---
 1 | 3 | 1
 2 | 4 | 4
(2 rows)

4.2.3 唯一索引推断

指定列名为c，但是在c列上无主键或者唯一约束，报错。

gaussdb=# INSERT INTO dst VALUES(1,2,3),(2,3,4) ON CONFLICT(c) DO UPDATE SET b = EXCLUDED.b;
ERROR:  There is no unique constraint matching the ON CONFLICT specification
gaussdb=# INSERT INTO dst VALUES(1,2,3),(2,3,4) ON CONFLICT(a,c) DO UPDATE SET b = EXCLUDED.b;
ERROR:  There is no unique constraint matching the ON CONFLICT specification

4.2.3 WHERE 条件

WHERE C > 3 只有在原表c列的值大于3时才进行更新。(2,4,4)满足条件被更新。

gaussdb=# INSERT INTO dst VALUES(1,10,3),(2,11,2) ON CONFLICT ON CONSTRAINT primary_dst DO UPDATE SET b = EXCLUDED.b WHERE c > 3;
INSERT 0 1
gaussdb=# SELECT * FROM dst;
 a | b  | c
---+----+---
 1 |  3 | 1
 2 | 11 | 4
(2 rows)

4.3 INSERT INTO SELECT

支持从源表中导入数据。

两条冲突数据更新b、c列，一条无冲突数据进行插入。

gaussdb=# SELECT * FROM src;
 a | b | c
---+---+---
 3 | 3 | 3
 1 | 1 | 1
 2 | 2 | 2
(3 rows)
gaussdb=# INSERT INTO dst SELECT * FROM src ON CONFLICT(a) DO UPDATE SET b = EXCLUDED.b + 4, c = EXCLUDED.c + 1;
INSERT 0 3
gaussdb=# SELECT * FROM dst ORDER BY 1;
 a | b | c
---+---+---
 1 | 5 | 2  -- 更新b、c列
 2 | 6 | 3  -- 更新b、c列
 3 | 3 | 3  -- 插入
(3 rows)

4.4 别名引用

建立一个名为EXCLUDED表，与UPDATE后面引用插入列方式EXCLUDED相冲突。可以使用 AS 语法指定别名避免冲突。

gaussdb=# CREATE TABLE EXCLUDED(d int unique, e int, f int);
gaussdb=# INSERT INTO EXCLUDED VALUES(1,2,3) ON CONFLICT(d) DO UPDATE SET e = EXCLUDED.e;
ERROR:  table reference "excluded" is ambiguous
LINE 1: ...ED VALUES(1,2,3) ON CONFLICT(d) DO UPDATE SET e = EXCLUDED.e...
                                                             ^
CONTEXT:  referenced column: e

gaussdb=# INSERT INTO EXCLUDED AS tt VALUES(1,2,3) ON CONFLICT(d) DO UPDATE SET e = EXCLUDED.e;

4.5 查看计划

显示推断主键或者唯一索引名称：

gaussdb=# EXPLAIN VERBOSE INSERT INTO dst VALUES(1,10,3),(2,11,2) ON CONFLICT ON CONSTRAINT primary_dst DO UPDATE SET b = EXCLUDED.b;
                                             QUERY PLAN
-----------------------------------------------------------------------------------------------------
  id |                operation                | E-rows | E-distinct | E-memory | E-width | E-costs
 ----+-----------------------------------------+--------+------------+----------+---------+---------
   1 | ->  Streaming (type: GATHER)            |      1 |            |          |      12 | 2.21
   2 |    ->  Insert on public.dst             |      4 |            |          |      12 | 2.11
   3 |       ->  Streaming(type: REDISTRIBUTE) |      4 |            | 2MB      |      12 | 0.11
   4 |          ->  Values Scan on "*VALUES*"  |      4 |            | 1MB      |      12 | 0.03

 Predicate Information (identified by plan id)
 ---------------------------------------------
   2 --Insert on public.dst
         Conflict Resolution: UPDATE
         Conflict Arbiter Indexes: primary_dst

EXPLAIN PERFORMANCE显示实际插入(Tuple Inserted) 与冲突更新(Tuple Conflicted)的数据条数。

gaussdb=# SELECT * FROM dst;
 a | b | c
---+---+---
 1 | 3 | 1
 2 | 4 | 2
 3 | 3 | 3
(3 rows)

gaussdb=# EXPLAIN PERFORMANCE INSERT INTO dst VALUES(1,10,3),(2,11,2),(5,1,1) ON CONFLICT ON CONSTRAINT primary_dst DO UPDATE SET b = EXCLUDED.b;

                                                                       QUERY PLAN
-----------------------------------------------------------------------------------------------------------------------------------------------------------
  id |                operation                |     A-time      | A-rows | E-rows | E-distinct |  Peak Memory   | E-memory | A-width | E-width | E-costs
 ----+-----------------------------------------+-----------------+--------+--------+------------+----------------+----------+---------+---------+---------
   1 | ->  Streaming (type: GATHER)            | 49.832          |      0 |      1 |            | 68KB           |          |         |      12 | 2.26
   2 |    ->  Insert on public.dst             | [5.391, 43.084] |      3 |      6 |            | [175KB, 176KB] |          |         |      12 | 2.17
   3 |       ->  Streaming(type: REDISTRIBUTE) | [5.387, 42.842] |      3 |      6 |            | [27KB, 28KB]   | 2MB      |         |      12 | 0.16
   4 |          ->  Values Scan on "*VALUES*"  | [0.013, 0.013]  |      3 |      6 |            | [6KB, 6KB]     | 1MB      |         |      12 | 0.04

 Predicate Information (identified by plan id)
 ----------------------------------------------
   2 --Insert on public.dst
          Conflict Resolution: UPDATE
          Conflict Arbiter Indexes: primary_dst
          Tuple Inserted: 1
          Tuple Conflicted: 2

5. 并发更新

在UPSERT概念中UPSERT可以原子性的执行如下：插入数据时判断是否有主键或者唯一索引冲突，如果无冲突，就进行插入，否则基于已有数据进行更新。在实现中可以保证UPSERT支持并发更新。有两个基本场景：

  • 两个线程同时执行UPSERT，且插入的数据在目标表中不存在并且互相冲突。

    两个线程各自都判断没有冲突，即将进行插入。但是只有一个线程会先拿到锁并插入。另一个线程在插入索引时发现索引冲突，那本次INSERT数据不能提交，需要在已有的数据上进行更新。

  • 两个线程同时执行UPSERT，且插入的数据在目标表已存在并且互相冲突。

    两个线程各自都判断有冲突，即将进行更新。一个线程拿到锁更新后，另一个需要在新数据上再次更新。但是拿到的老数据。所以需要再次扫描拿到最新的数据。

上面两个场景的简单描述相信大家可以理解UPSERT并发实现的原理了。

6. UPSERT约束

这里汇总下UPSERT的使用约束，部分已经在上文描述。

语法

  • 不支持与WITH子句同时使用。
  • 不支持与INSERT OVERWRITE同时使用。
  • UPDATE子句中VALUES(colname)用法不支持外层嵌套函数，即不支持类似sqrt(VALUES(colname))用法。如需支持，使用EXCLUDED.colname语法。
  • INSERT INTO ON CONFLICT(…) DO UPDATE必须有conflict_target。即必须指定列或者约束名。

功能

  • 不能更新分布列

  • 不支持对包含触发器(触发事件为INSERT或UPDATE)的目标表执行UPSERT语句。

  • 执行INSERT UPDATE语句时目标表必须有主键或者唯一索引。

  • UPDATE子句、UPDATE WHERE过滤条件、唯一索引推断处的条件表达式不支持不下推函数或表达式，否则报错。

  • UPDATE子句、UPDATE WHERE过滤条件不支持子查询，否则报错。

  • 当在列存表上执行UPSERT时，建议开启DELTA表，若DELTA表关闭会导致并发性能较差和空间膨胀的问题。

  • 列存复制表不支持通过UPSERT语句执行冲突更新操作。

  • 通过INSERT INTO SELECT语句执行UPSERT的更新操作时，需要注意SELECT语句的查询结果顺序。在分布式环境中未使用ORDER BY语句时每次执行相同的SELECT语句返回结果顺序可能不一样，这会导致UPSERT语句的执行结果不符合预期。

总结

GaussDB(DWS)提供的数据导入的功能UPSERT，对于实时数仓是一项非常关键的功能。在主键或者唯一索引冲突时能够执行更新或者忽略操作。简化业务层导入数据的逻辑。

想了解GaussDB(DWS)更多信息，欢迎微信搜索“GaussDB DWS”关注微信公众号，和您分享最新最全的PB级数仓黑科技，后台还可获取众多学习资料哦~

【版权声明】本文为华为云社区用户原创内容，转载时必须标注文章的来源（华为云社区）、文章链接、文章作者等基本信息， 否则作者和本社区有权追究责任。如果您发现本社区中有涉嫌抄袭的内容，欢迎发送邮件进行举报，并提供相关证据，一经查实，本社区将立刻删除涉嫌侵权内容，举报邮箱： cloudbbs@huaweicloud.com
EI企业智能 Gauss AP SQL 云数据仓库 GaussDB(DWS)
  • 点赞
  • 收藏
  • 关注作者

评论（0）

0/1000
抱歉，系统识别当前为高风险访问，暂不支持该操作

全部回复

上滑加载中

设置昵称

在此一键设置昵称，即可参与社区互动！

*长度不超过10个汉字或20个英文字符，设置后3个月内不可修改。

*长度不超过10个汉字或20个英文字符，设置后3个月内不可修改。

确认 取消

加入云驻计划，成为创作者

  • 华为云周边好礼
  • 免费体验产品
  • 特殊身份标识
  • 线下官方门票
  • 内部专家零距离
  • 与10000+优质创作者共同成长
立即加入