在 postgres plpgsql 函数中重用 json 解析的输入

Question

我有一个 plpgsql 函数，它接受 jsonb 输入，并使用它首先检查某些内容，然后再次在查询中获取结果。比如：

CREATE OR REPLACE FUNCTION public.my_func(
    a jsonb,
    OUT inserted integer)
    RETURNS integer
    LANGUAGE 'plpgsql'
    COST 100.0
    VOLATILE NOT LEAKPROOF
AS $function$
BEGIN
    -- fail if there's something already there 
    IF EXISTS(
    select t.x from jsonb_populate_recordset(null::my_type, a) f inner join some_table t
    on f.x = t.x and
       f.y = t.y
    ) THEN
    RAISE EXCEPTION 'concurrency violation... already present.';
    END IF;

    -- straight insert, and collect number of inserted
    WITH inserted_rows AS (
        INSERT INTO some_table (x, y, z)
        SELECT f.x, f.y, f.z
        FROM jsonb_populate_recordset(null::my_type, a) f
        RETURNING 1
    )
    SELECT count(*) from inserted_rows INTO inserted
    ;
END

在这里，我在IF 检查和实际插入中都使用了jsonb_populate_recordset(null::my_type, a)。有没有办法进行一次解析 - 也许通过某种变量？或者查询优化器是否会启动并确保解析操作只发生一次？

Answer 1

如果我理解正确，你会看到这样的东西：

CREATE OR REPLACE FUNCTION public.my_func(
    a jsonb,
    OUT inserted integer)
    RETURNS integer
    LANGUAGE 'plpgsql'
    COST 100.0
    VOLATILE NOT LEAKPROOF
AS $function$
BEGIN
    WITH checked_rows AS (
        SELECT f.x, f.y, f.z, t.x IS NOT NULL as present
        FROM jsonb_populate_recordset(null::my_type, a) f
        LEFT join some_table t
            on f.x = t.x and f.y = t.y
    ), vioalted_rows AS (
        SELECT count(*) AS violated FROM checked_rows AS c WHERE c.present
    ), inserted_rows AS (
        INSERT INTO some_table (x, y, z)
        SELECT c.x, c.y, c.z
        FROM checked_rows AS c
        WHERE (SELECT violated FROM vioalted_rows) = 0
        RETURNING 1
    )
    SELECT count(*) from inserted_rows INTO inserted
    ;

    IF inserted = 0 THEN 
        RAISE EXCEPTION 'concurrency violation... already present.';
    END IF;

END;
$function$;

Answer 2

JSONB 类型在赋值时不需要解析多次：

虽然 jsonb 数据以分解的二进制格式存储，但由于增加了转换开销，因此输入速度稍慢，但处理速度明显加快，因为不需要重新解析。

Link

jsonb_populate_recordset 函数声明为STABLE:

STABLE 表示该函数不能修改数据库，并且在单个表扫描中，对于相同的参数值，它将始终返回相同的结果，但其结果可能会在 SQL 语句中发生变化。

Link

我不确定。一方面 UDF 调用被视为单个语句，另一方面 UDF 可以包含多个语句。需要澄清。

最后，如果你想缓存这样的歌曲，那么你可以使用数组：

CREATE OR REPLACE FUNCTION public.my_func(
    a jsonb,
    OUT inserted integer)
    RETURNS integer
    LANGUAGE 'plpgsql'
    COST 100.0
    VOLATILE NOT LEAKPROOF
AS $function$
DECLARE
    d my_type[]; -- There is variable for caching 
BEGIN
    select array_agg(f) into d from jsonb_populate_recordset(null::my_type, a) as f;
    -- fail if there's something already there 
    IF EXISTS(
      select *
      from some_table t
      where (t.x, t.y) in (select x, y from unnest(d)))
    THEN
      RAISE EXCEPTION 'concurrency violation... already present.';
    END IF;

    -- straight insert, and collect number of inserted
    WITH inserted_rows AS (
        INSERT INTO some_table (x, y, z)
        SELECT f.x, f.y, f.z
        FROM unnest(d) f
        RETURNING 1
    )
    SELECT count(*) from inserted_rows INTO inserted;
END $function$;

Answer 3

如果你真的想重复重复使用一个结果set，一般的解决方案是临时表。示例：

• Using temp table in PL/pgSQL procedure for cleaning tables

但是，这相当昂贵。看起来您只需要一个 UNIQUE 约束或索引：

简单而安全的UNIQUE约束

ALTER TABLE some_table ADD CONSTRAINT some_table_x_y_uni UNIQUE (x,y);

与您的程序尝试相反，这也是并发安全的（没有竞争条件）。也快得多。

那么函数可以很简单：

CREATE OR REPLACE FUNCTION public.my_func(a jsonb, OUT inserted integer) AS
$func$
BEGIN
   INSERT INTO some_table (x, y, z)
   SELECT f.x, f.y, f.z
   FROM   jsonb_populate_recordset(null::my_type, a) f;

   GET DIAGNOSTICS inserted = ROW_COUNT;  -- OUT param, we're done here
END
$func$  LANGUAGE plpgsql;

如果(x,y) 已经存在于some_table 中，您将获得例外。为约束选择一个有指导意义的名称，在错误消息中报告。

我们可以用GET DIAGNOSTICS 读取命令标签，这比运行另一个计数查询便宜得多。

相关：

• How does PostgreSQL enforce the UNIQUE constraint / what type of index does it use?

UNIQUE 约束不可能？

对于 UNIQUE 约束不可行的不太可能的情况，您仍然可以让它相当简单：

CREATE OR REPLACE FUNCTION public.my_func(a jsonb, OUT inserted integer) AS
$func$
BEGIN
   INSERT INTO some_table (x, y, z)
   SELECT f.x, f.y, f.z  -- empty result set if there are any violations
   FROM  (
      SELECT f.x, f.y, f.z, count(t.x) OVER () AS conflicts
      FROM   jsonb_populate_recordset(null::my_type, a) f
      LEFT   JOIN some_table t USING (x,y)
      ) f
   WHERE  f.conflicts = 0;

   GET DIAGNOSTICS inserted = ROW_COUNT;

   IF inserted = 0 THEN
      RAISE EXCEPTION 'concurrency violation... already present.';
   END IF;

END
$func$  LANGUAGE plpgsql;

计算同一查询中的违规次数。 （count() 只计算非空值）。相关：

• Best way to get result count before LIMIT was applied

无论如何，您至少应该在some_table (x,y) 上有一个简单的索引。

重要的是要知道 plpgsql 在控制退出函数之前不会返回结果。异常取消返回，用户永远不会得到结果，只有错误消息。 We added a code example to the manual.

但是请注意，在并发写入负载下存在竞争条件。相关：

• Is SELECT or INSERT in a function prone to race conditions?

查询规划器会避免重复评估吗？

肯定不是在多个 SQL 语句之间。

即使函数本身定义为STABLE 或IMMUTABLE（示例中的jsonb_populate_recordset() 为STABLE），查询规划器也不知道输入参数的值在调用之间没有变化。跟踪并确保它会很昂贵。
实际上，由于 plpgsql 将 SQL 语句视为准备好的语句，这显然是不可能的，因为查询是计划的 before 参数值被提供给计划的查询。