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

Question

上下文

我们在 Postgres 数据库中进行基于模式的 Multi-Tenancy 。每个架构都与不同的租户相关联，并且具有完全相同的结构，除了一个名为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 语句的最佳方式是什么？

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

当前潜在的解决方案

到目前为止，似乎主要有两种可能有趣的方法(尽管我不太确定如何并行化查询执行):

1. 弄清楚如何在单个存储过程中执行所有操作，并使用 psql -c 简单地调用该 SP。
2. 使用 psql -c "SELECTtenant_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 或更早版本中，使用函数或 do block 而不是过程。

---

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

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);
...

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

---

以下函数发出通知，其中包含每个租户的已删除行数，并返回一个 block 的已删除行总数:

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);