如何在函数内部强制 COMMIT 以便其他会话可以看到更新的行？

Posted 2023-02-19

【中文标题】如何在函数内部强制 COMMIT 以便其他会话可以看到更新的行？

【英文标题】：How to force COMMIT inside function so other sessions can see updated row?

【发布时间】：2021-05-29 04:02:41

【问题描述】：

在 Postgres 12 数据库中，我在一个函数中有多个查询（SELECT、UPDATE、...），这些查询总共需要大约 20 分钟才能完成。 如果status 没有运行，我在顶部有一个检查UPDATE：

create or replace function aaa.fnc_work() returns varchar as 
$body$
    begin
        if (select count(*) from aaa.monitor where id='invoicing' and status='running')=0 then
           return 'running';
        else
           update aaa.monitor set status='running' where id='invoicing';
        end if;
        --- rest of code ---
        --finally
        update aaa.monitor set status='idle' where id='invoicing';
        return '';
    exception when others then
         return SQLERRM::varchar;
    end
$body$
language plpgsql;

这个想法是防止其他用户执行--- rest of code ---，直到status 空闲。

但是，其他人（调用相同的函数）似乎看不到更新的状态，他们也继续开始执行--- rest of code ---。之后如何强制提交：

更新 aaa.monitor set status='running' where id='invoicing';

以便所有其他用户会话可以看到更新后的status 并相应地退出。

我需要交易吗？

【参考方案1】：

您想要完成的是一个自治事务。 PostgreSQL 没有简单的方法来做到这一点。此链接here 讨论了一些替代方案。

但是，在上面链接的文章中讨论的一种方法是使用 PostgreSQL dblink 扩展。

您需要将扩展​​添加到服务器

CREATE EXTENSION dblink; 

然后你可以创建一个在你的函数中调用的新函数

CREATE FUNCTION update_monitor_via_dblink(msg text)
 RETURNS void
 LANGUAGE sql
AS $function$
   select dblink('host=/var/run/postgresql port=5432 user=postgres dbname=postgres',
    format(' update aaa.monitor set status= %M',msg::text)
$function$;

您可能要考虑的另一件事是使用 PostgreSQL 锁。更多信息可以找到here。

【讨论】：

谢谢@Ed Mendez。解决方案很好记

【参考方案2】：

继续阅读。我把最好的留到最后。

PROCEDURE 的概念证明

Postgres FUNCTION 始终是原子的（在单个事务包装器内运行）并且无法处理事务。所以 COMMIT 是不允许的。您可以使用dblink 的技巧来解决此问题。见：

Does Postgres support nested or autonomous transactions? How do I do large non-blocking updates in PostgreSQL?

但对于像这样的嵌套事务，请考虑使用 PROCEDURE。 Postgres 11 引入。您可以在那里管理交易：

CREATE OR REPLACE PROCEDURE aaa.proc_work(_id text, INOUT _result text = NULL)
  LANGUAGE plpgsql AS
$proc$
BEGIN
   -- optionally assert that the steering row exists
   PERFORM FROM aaa.monitor WHERE id = _id FOR KEY SHARE SKIP LOCKED;
   IF NOT FOUND THEN   
      RAISE EXCEPTION 'aaa.monitor.id = % not found or blocked!', quote_literal(_id);
   END IF;

   -- try UPDATE
   UPDATE aaa.monitor
   SET    status = 'running'
   WHERE  id = _id
   AND    status <> 'running';  -- assuming column is NOT NULL

   IF NOT FOUND THEN
      _result := 'running'; RETURN;  -- this is how you return with INOUT params
   END IF;

   COMMIT;                                   -- HERE !!!

   <<big_work>>   -- optional label for the nested block
   BEGIN          -- start new code block
      --- rest of code ---
      -- PERFORM 1/0;          -- test exception?
      -- PERFORM pg_sleep(5);  -- test concurrency?

      -- finally
      UPDATE aaa.monitor
      SET    status = 'idle'
      WHERE  id = _id;
      _result := ''; RETURN;

   EXCEPTION WHEN OTHERS THEN
      UPDATE aaa.monitor
      SET    status = 'idle'  -- reset!
      WHERE  id = _id;
      _result := SQLERRM;
   END big_work;
END
$proc$;

打电话（重要！）：

CALL aaa.proc_work('invoicing');  -- stand-alone call!

重要提示

我在UPDATE 之后添加了COMMIT。现在，并发事务可以看到更新的行。

但是没有额外的BEGIN 或START TRANSACTION。 The manual:

在由CALL 命令以及匿名代码调用的过程中 块（DO 命令），可以使用 命令COMMIT 和ROLLBACK。新事务开始 使用这些命令结束事务后自动执行，因此 没有单独的START TRANSACTION 命令。 （注意BEGIN 和 END 在 PL/pgSQL 中有不同的含义。）

我们需要一个单独的 PL/pgSQL code block，因为您有一个自定义异常处理程序，并且（引用 the manual）：

事务不能在带有异常处理程序的块内结束。

您不能在另一个事务中调用此过程，也不能与任何其他 DML 语句一起调用，这将强制外部事务包装器。必须是独立的CALL。见：

Unable to Create COMMIT inside PostgreSQL 11.5 Procedure

注意异常处理程序中添加的UPDATE aaa.monitor SET status = 'idle' WHERE ...。否则（已提交！）status 将在异常发生后无限期地保持“运行”。

关于从过程中返回值：

How to return a value from a stored procedure (not function)?

我在INOUT 参数中添加了DEFAULT NULL，因此您不必在调用时提供参数。

UPDATE 直接。如果该行正在“运行”，则不会发生更新。 （这也修正了逻辑：您的 IF 表达式似乎是倒退的，因为当找到带有 status='running' 的 no 行时它返回“正在运行”。似乎您想要相反的结果。）

我添加了一个（可选！）断言以确保表 aaa.monitor 中的行存在。添加FOR KEY SHARE 锁也消除了断言和以下UPDATE 之间的竞争条件的微小时间窗口。锁与删除或更新 PK 列冲突 - 但不 与更新 status。所以在正常操作中永远不会引发异常！ The manual:

目前，UPDATE 案例考虑的列集是 那些有唯一索引的，可以在国外使用的 键（因此不考虑部分索引和表达式索引）， 但这可能会在未来发生变化。

SKIP LOCK 在发生锁冲突的情况下不等待。永远不会发生添加的异常。只是展示了一个无懈可击的概念证明。

您的更新显示aaa.monitor 中有25 行，所以我添加了参数_id。

优越的方法

以上内容可能有助于保留更多信息以供全世界查看。对于队列操作，有更有效的解决方案。使用 lock 代替，它对其他人立即“可见”。那么你就不需要嵌套事务了，一个普通的FUNCTION 就可以了：

CREATE OR REPLACE FUNCTION aaa.fnc_work(_id text)
  RETURNS text
  LANGUAGE plpgsql AS
$func$
BEGIN
   -- optionally assert that the steering row exists
   PERFORM FROM aaa.monitor WHERE id = _id FOR KEY SHARE SKIP LOCKED;
   IF NOT FOUND THEN   
      RAISE EXCEPTION 'aaa.monitor.id = % not found or blocked!', quote_literal(_id);
   END IF;

   -- lock row
   PERFORM FROM aaa.monitor WHERE id = _id FOR NO KEY UPDATE SKIP LOCKED;

   IF NOT FOUND THEN
      -- we made sure the row exists, so it must be locked
      RETURN 'running';
   END IF;

   --- rest of code ---
   -- PERFORM 1/0;          -- test exception?
   -- PERFORM pg_sleep(5);  -- test concurrency?

   RETURN '';

EXCEPTION WHEN OTHERS THEN
   RETURN SQLERRM;

END
$func$;

呼叫：

SELECT aaa.fnc_work('invoicing');

调用可以以任何你想要的方式嵌套。只要一项事务在处理大工作，就不会启动其他事务。

同样，可选的断言取出FOR KEY SHARE 锁以消除竞争条件的时间窗口，并且在正常操作中绝不应发生添加的异常。

我们根本不需要status 列。行锁本身就是看门人。因此PERFORM FROM aaa.monitor ... 中的SELECT 列表为空。附带好处：这也不会通过来回更新行来产生死元组。如果由于其他原因您仍需要更新status，您将回到上一章的可见性问题。您可以将两者结合起来......

关于PERFORM：

SELECT or PERFORM in a PL/pgSQL function

关于行锁：

Postgres UPDATE … LIMIT 1

【讨论】：

非常感谢您的精彩回答。我已经注意到我拥有的所有选项。我想采用高级方法。在我最初的帖子中，我错过了where id='invoicing'，因为 aa.monitor 有多行。我可以问一下，对于多行，我仍然可以使用锁定，因为我将更新同一条记录吗？这种情况下status 到“正在运行”的更新在哪里？我不认为它是锁定方法解决方案

@KofiB：我适应了多个 ID。 “高级方法”同样有效。没有实际的UPDATE，只是一个行锁。这就是它的美妙之处——提高效率和知名度。

@KofiB：我更新了两个要点：1. 消除程序和函数中竞争条件的极端情况。 2. 直接在程序中UPDATE，这样更快。