对 Postgres 中的所有模式运行查询

Question

上下文

我们在 Postgres 数据库中进行基于模式的多租户。每个架构都与不同的租户相关联，并且将具有完全相同的结构除了一个名为 public 的架构。

要获取所有相关模式的列表，我们可以使用：

SELECT tenant_schema FROM public.tenant_schema_mappings;

问题

我们需要定期清理所有租户的特定表：

DELETE FROM a_tenant_schema.parent WHERE expiration_date_time < NOW(); 

（实际查询要复杂一些，因为我们还需要删除与parent 关联的链接children 条目，但为了这个问题，让我们保持简单。）

约束

1. 我们无法使用 pg_cron，因为我们的数据库服务器托管在 Azure 上，并且尚不支持该扩展。
2. 我们不希望仅仅为了执行 cron 作业而部署整个服务/应用程序。
3. 因此，我们决定使用部署在 k8s 命名空间中的 CronJob pod，从而可以通过 shell 命令使用 psql 客户端直接与数据库通信。

问题

在 shell 中使用 psql 对所有相关架构执行给定的 DELETE 语句的最佳方法是什么？

请记住：由于可能有数百个租户，因此为每个租户并行运行清理查询可能会很有趣。

当前的潜在解决方案

到目前为止，似乎主要有 2 种方法可能很有趣（尽管我不太确定如何并行化查询执行）：

1. 弄清楚如何在单个存储过程中执行所有操作，然后使用 psql -c 简单地调用该 SP。
2. 使用psql -c "SELECT tenant_schema FROM public.tenant_schema_mappings;" 收集所有相关租户模式的列表，然后使用shell 命令通过动态构造适当的查询来遍历该列表。使用查询结果集，使用psql -c 一个一个地运行它们。

其他部分解决方案

我认为我们实际上可以使用以下 SQL 构造查询：

SELECT 'DELETE * FROM ' || tenant_schema || '.parent WHERE expiration_date_time < NOW();' AS query
FROM public.tenant_schema_mappings;

也许有办法告诉 Postgres 执行所有结果字符串？

Answer 1

您可以定义一个使用dynamic commands, 的 Postgres 过程，例如：

create or replace procedure clear_tenants()
language plpgsql as $function$
declare
    tenant text;
begin
    for tenant in
        select tenant_schema 
        from public.tenant_schema_mappings
    loop
        execute format($ex$
            delete from %I.parent 
            where expiration_date_time < now()
            $ex$, tenant);
    end loop;
end
$function$;

那么你的 cron 任务所要做的就是调用这个过程：

call clear_tenants()

在 Postgres 10 或更早版本中，使用函数或执行块而不是过程。

---

这个简单解决方案的主要缺点是所有内容都在单个事务中执行。不幸的是，您无法控制包含动态查询的过程中的事务。我会在描述模式的表中定义chunk_number，并在其自己的事务中为每个块调用该过程。

create or replace procedure public.clear_tenants(chunk integer)
language plpgsql as $function$
declare
    tenant text;
begin
    for tenant in
        select tenant_schema 
        from public.tenant_schema_mappings
        where chunk_number = chunk
    loop
        execute format($ex$
            delete from %I.parent 
            where expiration_date_time < now()
            $ex$, tenant);
    end loop;
end
$function$;

在客户端，我必须准备一个格式如下的脚本：

-- in psql the procedures will be executed in separate transactions
-- if you do not use begin; explicitly
call clear_tenants(1);
call clear_tenants(2);
call clear_tenants(3);
...

或为单个块执行许多 psql 实例（每个都在一个单独的连接中）。最后一个选项实际上是强制并发的唯一方法。当然，它受到合理数量的并发连接的限制。

---

以下函数从每个租户发出包含已删除行数的通知，并返回块的已删除行总数：

create or replace function public.clear_tenants_modified(chunk integer)
returns bigint language plpgsql as $function$
declare
    tenant text;
    deleted_rows bigint;
    total_deleted_rows bigint = 0;
begin
    for tenant in
        select tenant_schema 
        from public.tenant_schema_mappings
        where chunk_number = chunk
    loop
        execute format($ex$
            with delete_statement as (
                delete from %I.parent 
                where expiration_date_time < now()
                returning 1 as x)
            select count(x)
            from delete_statement
            $ex$, tenant)
        into deleted_rows;
        raise notice '%: %', tenant, deleted_rows;
        total_deleted_rows = total_deleted_rows+ deleted_rows;
    end loop;
    return total_deleted_rows;
end
$function$;

select clear_tenants_modified(1);