声明并返回 DELETE 和 INSERT 的值

Question

我正在尝试根据唯一 ID 从我们的某些数据库中删除重复数据。所有已删除的数据应存储在单独的表中以供审核。由于它涉及相当多的数据库以及不同的模式和表，我想开始使用变量来减少出错的机会和我需要做的工作量。

这是我能想到的最好的示例查询，但它不起作用：

do $$
declare @source_schema  varchar := 'my_source_schema';
declare @source_table   varchar := 'my_source_table';
declare @target_table   varchar := 'my_target_schema' || source_table || '_duplicates'; --target schema and appendix are always the same, source_table is a variable input.
declare @unique_keys    varchar := ('1', '2', '3') 

begin 
select into @target_table
from @source_schema.@source_table
where id in (@unique_keys);

delete from @source_schema.@source_table where export_id in (@unique_keys);

end ;
$$;

查询语法适用于硬编码值。

大多数情况下，我的变量被视为列或根本无法识别。 :(

Answer 1

您需要创建并调用带有输入参数的 plpgsql 过程：

CREATE OR REPLACE PROCEDURE duplicates_suppress
(my_target_schema text, my_source_schema text, my_source_table text, unique_keys text[])
LANGUAGE plpgsql AS
$$
BEGIN

EXECUTE FORMAT(
'WITH list AS (INSERT INTO %1$I.%3$I_duplicates SELECT * FROM %2$I.%3$I WHERE array[id] <@ %4$L :: integer[] RETURNING id)
DELETE FROM %2$I.%3$I AS t USING list AS l WHERE t.id = l.id', my_target_schema, my_source_schema, my_source_table, unique_keys :: text) ;

END ;
$$ ;

过程duplicates_suppress 将my_source_schema.my_source_table 中id 在数组unique_keys 中的行插入my_target_schema.my_source_table || '_duplicates'，然后从表my_source_schema.my_source_table 中删除这些行。

在dbfiddle查看测试结果。

Answer 2

如前所述，您需要某种动态 SQL。在 FUNCTION、PROCEDURE 或 DO 语句中在服务器上执行此操作。

您应该对PL/pgSQL 感到满意。动态 SQL 不是初学者的玩具。

PROCEDURE 的示例，就像 Edouard 已经建议的那样。您将需要 FUNCTION 来将其包装在外部事务中（就像您很可能一样）。见：

• When to use stored procedure / user-defined function?

CREATE OR REPLACE PROCEDURE pg_temp.f_archive_dupes(_source_schema text, _source_table text, _unique_keys int[], OUT _row_count int)
  LANGUAGE plpgsql AS
$proc$
   -- target schema and appendix are always the same, source_table is a variable input
DECLARE
   _target_schema CONSTANT text := 's2';  -- hardcoded
   _target_table  text := _source_table || '_duplicates';
   _sql           text := format(
'WITH del AS (
   DELETE FROM %I.%I
   WHERE  id = ANY($1)
   RETURNING *
   )
INSERT INTO %I.%I TABLE del', _source_schema, _source_table
                            , _target_schema, _target_table);
BEGIN
   RAISE NOTICE '%', _sql;           -- debug
   EXECUTE _sql USING _unique_keys;  -- execute

   GET DIAGNOSTICS _row_count = ROW_COUNT;
END
$proc$;

呼叫：

CALL pg_temp.f_archive_dupes('s1', 't1', '{1, 3}', 0);

db小提琴here

我将程序设为临时，因为我认为您不需要永久保留它。每个数据库创建一次。见：

• How to create a temporary function in PostgreSQL?

传递的架构和表名是区分大小写的字符串！ （与普通 SQL 中不带引号的标识符不同。）无论哪种方式，在动态连接 SQL 时都要小心 SQL 注入。见：

• Are PostgreSQL column names case-sensitive?
• Table name as a PostgreSQL function parameter

制作 _unique_keys 类型 int[]（integer 的数组），因为您的样本值看起来像整数。使用id 列的实际数据类型！

变量_sql 保存查询字符串，因此可以在实际执行之前轻松调试。为此目的使用RAISE NOTICE '%', _sql;。
我建议在您确定之前评论EXECUTE 行。

我让PROCEDURE 返回已处理的行数。您没有要求这样做，但它通常很方便。几乎不惜任何代价。见：

• Dynamic SQL (EXECUTE) as condition for IF statement
• Best way to get result count before LIMIT was applied

最后但同样重要的是，在数据修改 CTE 中使用 DELETE ... RETURNING *。因为只有在单独的SELECT 和DELETE 的成本大约一半 时才能找到行。而且它非常安全。如果出现任何问题，无论如何都会回滚整个事务。
两个单独的命令也可能遇到并发问题或竞争条件，因为DELETE 隐式锁定要删除的行。示例：

• Replicating data between Postgres DBs

---

或者您可以在客户端程序中构建语句。像 psql 一样，使用 \gexec。示例：

• Filter column names from existing table for SQL DDL statement

Answer 3

基于 Erwin 的回答，小优化...

create or replace procedure pg_temp.p_archive_dump
    (_source_schema text, _source_table text,
        _unique_key int[],_target_schema text)
language plpgsql as
    $$
    declare
        _row_count bigint;
        _target_table text := '';
    BEGIN
        select quote_ident(_source_table) ||'_'|| array_to_string(_unique_key,'_')   into _target_table from quote_ident(_source_table);
        raise notice 'the deleted table records will store in %.%',_target_schema, _target_table;
        execute format('create table %I.%I as select * from %I.%I limit 0',_target_schema, _target_table,_source_schema,_source_table );

        execute format('with mm as ( delete from %I.%I where id = any (%L) returning * ) insert into %I.%I table mm'

            ,_source_schema,_source_table,_unique_key, _target_schema, _target_table);
    GET DIAGNOSTICS _row_count = ROW_COUNT;
    RAISE notice 'rows influenced, %',_row_count;
    end
$$;

-- 如果您的 _unique_key 不是很多，此解决方案还会为您创建一个表。显然，您需要自己创建目标架构。
如果您的 unique_key 过多，您可以自定义以正确重命名转储表。
我们称之为。
call pg_temp.p_archive_dump('s1','t1', '{1,2}','s2');
s1 是源模式，t1 是源表，{1,2} 是您要提取到新表的唯一键。 s2 是目标架构