如何在更新链的末尾触发触发器？

Question

我有几个使用触发器相互交互的表,而我一直处理触发器执行的当前方式使用的pg_trigger_depth() < 2是丑陋的.我真的希望最终的触发器只运行一次,并且在所有的每行事件发生之后结束.不幸的是,CONSTRAINT TRIGGERs FOR EACH ROW只是,并且FOR STATEMENT触发器实际上在触发器中每个语句触发一次,而不是每个初始语句启动它.

我已经查看了围绕这个主题的其他几个SO问题,并且没有找到类似于我正在做的事情.

这是设置:

CREATE TABLE report(
  report_tk SERIAL PRIMARY KEY,
  report_id UUID NOT NULL,
  report_name TEXT NOT NULL,
  report_data INT NOT NULL,
  report_subscribers TEXT[] NOT NULL DEFAULT ARRAY[]::TEXT[],
  valid_range TSTZRANGE NOT NULL DEFAULT '(,)',
  EXCLUDE USING GIST ((report_id :: TEXT) WITH =, report_name WITH =, valid_range WITH &&)
);
CREATE TABLE report_subscriber(
  report_id INT NOT NULL REFERENCES report ON DELETE CASCADE;
  subscriber_name TEXT NOT NULL,
  needs_sync BOOLEAN NOT NULL DEFAULT TRUE,
  EXCLUDE USING GIST (subscriber_name WITH =, valid_range WITH &&)
);
CREATE OR REPLACE FUNCTION sync_subscribers_to_report()
  RETURNS TRIGGER LANGUAGE plpgsql SET SEARCH_PATH TO dwh, public AS $$
BEGIN
  RAISE INFO 'Running sync to report trigger';

  BEGIN
    CREATE TEMPORARY TABLE lock_sync_subscribers_to_report(
    ) ON COMMIT DROP;
    RAISE INFO 'syncing to report, stack depth is: %', pg_trigger_depth();
    UPDATE report r
    SET report_subscribers = x.subscribers
    FROM (
           SELECT
             report_tk
             , array_agg(DISTINCT u.subscriber_name ORDER BY u.subscriber_name) AS subscribers
           FROM report_subscriber s
           WHERE s.report_tk IN (
             SELECT DISTINCT report_tk
             FROM report_subscriber s2
             WHERE s.needs_sync
           )
           GROUP BY s.report_tk
         ) x
    WHERE r.report_tk = x.report_tk;
    RAISE INFO 'turning off sync flag, stack depth is: %', pg_trigger_depth();
    UPDATE report_subscriber
    SET needs_sync = FALSE
    WHERE needs_sync = TRUE;
    RETURN NULL;
  EXCEPTION WHEN DUPLICATE_TABLE THEN
    RAISE INFO 'skipping recursive call, stack depth is: %', pg_trigger_depth();
    RETURN NULL;
  END;
END;
$$;
CREATE TRIGGER sync_subscribers_to_report
  AFTER INSERT OR UPDATE OR DELETE
  ON report_subscriber
  FOR STATEMENT
EXECUTE PROCEDURE sync_subscribers_to_report();

因此,通过此设置,我希望能够:

• 插入报告记录
• 保证报告名称在任何单个时间点只能存在一次(在valid_range上的EXCLUDE)
• 在订阅者表中插入报告订阅者
• 保证订户不能一次订阅多个报告.
• 允许多人订阅报告.
• 只要将记录添加到订户表,请将该名称添加到报表中的订户列表中.
• 每当从订户表中删除记录时,从报告表中的订户列表中删除该名称.
• 每当从报告表中删除记录时,删除相应的订户记录(由...处理) ON DELETE CASCADE

如果在单个语句中对订阅者表进行了大量编辑(常见情况),则最好只运行一个简单查询,以使用订阅者表中新记录和剩余记录的聚合来更新报告表.

我的原始解决方案涉及向needs_update订户表添加一个标志并触发该标志以进行更新然后关闭标志.当然,这导致触发器的另一次触发,我停止了pg_trigger_depth() < 2(2是因为插入可能是由系统中的其他触发器引起的).除了丑陋之外,触发器功能中的语句还会导致更多的FOR EACH STATEMENT触发发生,这也很烦人.

我尝试使用我在其他SO答案(/sf/answers/626544761/)中创建临时表并捕获欺骗表异常以防止进一步执行的技巧的不同版本的标志.我认为这并没有真正改善这个问题.

有没有办法以干净的方式做我想做的事情？虽然这是一个明显的玩具示例,但我的真实应用程序确实需要构建数据的"打包数组"表示,并且以有效的方式这样做会很棒.

Answer 1

report_subscriber我认为你最好使用一个单独的挂起更改队列,而不是自己使用一个标志.这有一些好处:

• 没有触发器递归
• 在引擎盖下,UPDATE只是DELETE+重新INSERT,所以插入队列实际上比翻转旗帜便宜
• 可能要便宜得多,因为你只需要对不同的report_ids 进行排队,而不是克隆整个report_subscriber记录,你可以在临时表中进行,因此存储是连续的,不需要同步到磁盘
• 翻转标志时没有竞争条件,因为队列是当前事务的本地队列(在您的实现中,受此影响的记录UPDATE report_subscriber不一定与您在SELECT...中记录的记录相同)

因此,初始化队列表:

CREATE FUNCTION create_queue_table() RETURNS TRIGGER LANGUAGE plpgsql AS $$
BEGIN
  CREATE TEMP TABLE pending_subscriber_changes(report_id INT UNIQUE) ON COMMIT DROP;
  RETURN NULL;
END
$$;

CREATE TRIGGER create_queue_table_if_not_exists
  BEFORE INSERT OR UPDATE OF report_id, subscriber_name OR DELETE
  ON report_subscriber
  FOR EACH STATEMENT
  WHEN (to_regclass('pending_subscriber_changes') IS NULL)
  EXECUTE PROCEDURE create_queue_table();

...在到达时排队更改,忽略已经排队的任何内容:

CREATE FUNCTION queue_subscriber_change() RETURNS TRIGGER LANGUAGE plpgsql AS $$
BEGIN
  IF TG_OP IN ('DELETE', 'UPDATE') THEN
    INSERT INTO pending_subscriber_changes (report_id) VALUES (old.report_id)
    ON CONFLICT DO NOTHING;
  END IF;

  IF TG_OP IN ('INSERT', 'UPDATE') THEN
    INSERT INTO pending_subscriber_changes (report_id) VALUES (new.report_id)
    ON CONFLICT DO NOTHING;
  END IF;
  RETURN NULL;
END
$$;

CREATE TRIGGER queue_subscriber_change
  AFTER INSERT OR UPDATE OF report_id, subscriber_name OR DELETE
  ON report_subscriber
  FOR EACH ROW
  EXECUTE PROCEDURE queue_subscriber_change();

...并在语句末尾处理队列:

CREATE FUNCTION process_pending_changes() RETURNS TRIGGER LANGUAGE plpgsql AS $$
BEGIN
  UPDATE report
  SET report_subscribers = ARRAY(
    SELECT DISTINCT subscriber_name
    FROM report_subscriber s
    WHERE s.report_id = report.report_id
    ORDER BY subscriber_name
  )
  FROM pending_subscriber_changes c
  WHERE report.report_id = c.report_id;

  DROP TABLE pending_subscriber_changes;
  RETURN NULL;
END
$$;

CREATE TRIGGER process_pending_changes
  AFTER INSERT OR UPDATE OF report_id, subscriber_name OR DELETE
  ON report_subscriber
  FOR EACH STATEMENT
  EXECUTE PROCEDURE process_pending_changes();

这有一个小问题:UPDATE不提供有关更新订单的任何保证.这意味着,如果这两个语句同时运行:

INSERT INTO report_subscriber (report_id, subscriber_name) VALUES (1, 'a'), (2, 'b');
INSERT INTO report_subscriber (report_id, subscriber_name) VALUES (2, 'x'), (1, 'y');

...如果他们试图以report相反的顺序更新记录,那么就有可能出现死锁.您可以通过对所有更新强制执行一致的排序来避免这种情况,但遗憾的是无法附加ORDER BY到UPDATE语句; 我认为你需要诉诸游标:

CREATE FUNCTION process_pending_changes() RETURNS TRIGGER LANGUAGE plpgsql AS $$
DECLARE
  target_report CURSOR FOR
    SELECT report_id
    FROM report
    WHERE report_id IN (TABLE pending_subscriber_changes)
    ORDER BY report_id
    FOR NO KEY UPDATE;
BEGIN
  FOR target_record IN target_report LOOP
    UPDATE report
    SET report_subscribers = ARRAY(
        SELECT DISTINCT subscriber_name
        FROM report_subscriber
        WHERE report_id = target_record.report_id
        ORDER BY subscriber_name
      )
    WHERE CURRENT OF target_report;
  END LOOP;

  DROP TABLE pending_subscriber_changes;
  RETURN NULL;
END
$$;

如果客户端尝试在同一事务中运行多个语句,则仍有可能死锁(因为更新排序仅在每个语句中应用,但更新锁保持到提交为止).您可以通过process_pending_changes()在事务结束时仅触发一次来解决此问题(缺点是,在该事务中,您将看不到自己的更改反映在report_subscribers数组中).

以下是"on commit"触发器的通用大纲,如果您认为填写它是值得的:

CREATE FUNCTION run_on_commit() RETURNS TRIGGER LANGUAGE plpgsql AS $$
BEGIN
  <your code goes here>
  RETURN NULL;
END
$$;

CREATE FUNCTION trigger_already_fired() RETURNS BOOLEAN LANGUAGE plpgsql VOLATILE AS $$
DECLARE
  already_fired BOOLEAN;
BEGIN
  already_fired := NULLIF(current_setting('my_vars.trigger_already_fired', TRUE), '');
  IF already_fired IS TRUE THEN
    RETURN TRUE;
  ELSE
    SET LOCAL my_vars.trigger_already_fired = TRUE;
    RETURN FALSE;
  END IF;
END
$$;

CREATE CONSTRAINT TRIGGER my_trigger
  AFTER INSERT OR UPDATE OR DELETE ON my_table
  DEFERRABLE INITIALLY DEFERRED
  FOR EACH ROW
  WHEN (NOT trigger_already_fired())
  EXECUTE PROCEDURE run_on_commit();