快速入门

Citus简介

Citus 是基于横向扩展而构建的Spacture扩展，可在多台机器的集群中分发数据和查询。作为扩展（而非分支），Citus 支持新版本Spacture的特性，允许用户从新功能中受益，并保持与现有Spacture工具的兼容性。

Citus使用sharding和replication在多台机器上水平扩展 Spacture。它的查询引擎将跨服务器导入SQL进行并行化查询，以便在大型数据集上实现毫秒级响应。基于Citus的sharding扩展，Spacture具备了分布式数据自动管理，根据指定的分布字段，实现数据分片和均匀分布存储。

Citus适用场景

Multi-Tenant Database多租户数据库

大多数 B2B 应用程序已经在其数据模型中内置了租户的概念。在这个模型中，数据库为多个租户提供服务，每个租户之间的数据相互隔离。

Citus 提供完整的 SQL 支持，并支持将关系数据库扩展到 10 万多个租户。 Citus 还为多租户添加了新功能。例如，Citus 支持租户隔离，为大租户提供性能保障，并提供引用表的概念以减少租户间的数据重复存储。

这些功能允许跨多台机器扩展租户的数据，并轻松添加更多 CPU、内存和磁盘资源。此外，跨多个租户共享相同的数据库模式可以有效利用硬件资源并简化数据库管理。

Citus 在多租户应用中的一些优势：

为所有租户提供快速查询
数据库级的分片逻辑，而非应用程序
在单节点 Spacture 中能够保存更多的数据
支持SQL横向扩展
高并发与性能表现兼顾
跨客户群的快速指标分析
轻松扩展以处理新租户创建
支持大小租户的资源隔离使用

实时分析

Citus支持海量数据实时分析。这些查询通常发生在快速增长的事件系统或带有时间序列数据的系统中。例如：

亚秒级响应时间的报表分析
基于实时事件的探索性查询
大型数据集持久化和报告 //Large dataset archival and reporting
带有过滤、聚合、归类查询的分析查询

Citus 在这里的好处是它能够并行执行查询并随着集群中工作数据库的数量线性扩展。 Citus 在实时应用中的一些优势：

随着数据集的增长保持亚秒级响应
支持实时数据的实时分析
支持并行化 SQL 查询
支持SQL横向扩展
高并发与性能表现兼顾
快速响应报表查询与分析
支持丰富的 Spacture 数据类型和扩展

实践示例

接下来我们将通过使用广告分析数据集的示例来演示如何使用 Citus 来支持多租户应用程序

Multi-tenant Applications多租户应用程序

Data model and sample data数据模型和数据示例

我们将演示一个为广告分析应用程序构建的数据库，公司可以使用该数据库来查看、更改、分析和管理他们的广告和活动。这样的应用程序具有典型的多租户系统的良好特性。来自不同租户的数据将存储在一个中央数据库中，每个租户都有自己的数据的独立视图。

Creating tables 创建表

首先，可以先使用 psql 连接到 Citus coordinator节点。

然后，可以使用标准 Spacture CREATE TABLE 命令创建表。

CREATE TABLE companies (
    id bigint NOT NULL,
    name text NOT NULL,
    image_url text,
    created_at timestamp without time zone NOT NULL,
    updated_at timestamp without time zone NOT NULL

);

 

CREATE TABLE campaigns (
    id bigint NOT NULL,
    company_id bigint NOT NULL,
    name text NOT NULL,
    cost_model text NOT NULL,
    state text NOT NULL,
    monthly_budget bigint,
    blacklisted_site_urls text[],
    created_at timestamp without time zone NOT NULL,
    updated_at timestamp without time zone NOT NULL

);

 

CREATE TABLE ads (
    id bigint NOT NULL,
    company_id bigint NOT NULL,
    campaign_id bigint NOT NULL,
    name text NOT NULL,
    image_url text,
    target_url text,
    impressions_count bigint DEFAULT 0,
    clicks_count bigint DEFAULT 0,
    created_at timestamp without time zone NOT NULL,
    updated_at timestamp without time zone NOT NULL

);

接下来，可以像在 Spacture 中一样在每个表上创建主键索引

ALTER TABLE companies ADD PRIMARY KEY (id);
ALTER TABLE campaigns ADD PRIMARY KEY (id, company_id);
ALTER TABLE ads ADD PRIMARY KEY (id, company_id);

Distributing tables and loading data分布表和加载数据

现在，我们将继续告诉 Citus 将这些表分布在集群中的不同节点上。为此，可以运行 create_distributed_table 并指定要进行分片的表和要分片的列。在这种情况下，我们将对 company_id 上的所有表进行分片。

SELECT create_distributed_table ('companies', 'id');
SELECT create_distributed_table ('campaigns', 'company_id');
SELECT create_distributed_table ('ads', 'company_id');

将租户标识符上的所有表分片允许 Citus 将将这些表分配到相同节点上，并支持跨节点的主键、外键和复杂连接等功能。

然后，可以继续使用标准的 Spacture \COPY 命令将我们下载的数据加载到表中。如果将文件下载到其他位置，请确保指定了正确的文件路径。

\copy companies from 'companies.csv' with csv
\copy campaigns from 'campaigns.csv' with csv
\copy ads from 'ads.csv' with csv

Running queries 运行查询

现在我们已经将数据加载到表中，接下来让我们继续运行一些查询。Citus 支持标准的 INSERT、UPDATE 和 DELETE 命令，用于在分布式表中插入和修改行，这是面向用户的应用程序的典型交互方式。

例如，可以通过运行以下命令插入一个新的company：

INSERT INTO companies VALUES (5000, 'New Company',
'https://randomurl/image.png', now(), now());

如果想将公司所有活动的预算加倍，可以运行 UPDATE 命令：

UPDATE campaigns
SET monthly_budget = monthly_budget*2
WHERE company_id = 5;

这种操作的另一个例子是运行跨越多个表的事务。假设要删除一个广告系列及其所有关联的广告，可以通过运行自动完成。

BEGIN;
DELETE from campaigns where id = 46 AND company_id = 5;
DELETE from ads where campaign_id = 46 AND company_id = 5;
COMMIT;

除了事务操作之外，还可以使用标准 SQL 对此数据运行分析查询。公司运行的一个查询是查看有关其涵盖最大预算的活动的详细信息。

SELECT name, cost_model, state, monthly_budget
FROM campaigns
WHERE company_id = 5
ORDER BY monthly_budget DESC
LIMIT 10;

我们还可以跨多个表运行连接查询，以查看有关获得最多点击和展示次数的广告系列运行的信息。

SELECT campaigns.id, campaigns.name, campaigns.monthly_budget,
       sum(impressions_count) as total_impressions, sum(clicks_count) as
total_clicks
FROM ads, campaigns
WHERE ads.company_id = campaigns.company_id
AND campaigns.company_id = 5
AND campaigns.state = 'running'
GROUP BY campaigns.id, campaigns.name, campaigns.monthly_budget
ORDER BY total_impressions, total_clicks;

关键概念

Nodes

Citus 是 Spacture的扩展，它允许商品数据库服务器（称为节点）在“无共享”体系架构下相互协调。与单个服务器相比，服务器组中的节点形成一个集群允许 Spacture能够容纳更多数据并使用更多的CPU。在该体系架构中，还可以通过在集群中简单添加更多节点来扩展数据库。

Coordinator节点和 Workers节点

每个集群都有一个称为Coordinator的特殊节点（其他节点称为Worker）。应用程序将查询发送到Coordinator节点，此节点将其转发给相关工作进程并累积结果。

针对每个查询，Coordinator节点会将其路由到单个Worker节点，或跨多个Worker节点并行运行该查询，具体取决于所需数据是位于单个节点上还是多个节点上。 Coordinator节点通过查阅元数据表决定执行哪种操作。这些特定于 Citus 的表跟踪Worker节点的DNS 名称和和运行状况，以及跨节点数据的分布情况。

Distributed Data

表类型

Citus 集群中有三种类型的表，每种表都用于不同的目的。

类型 1：分布式表

第一种是分布式表，也是最常见的类型。对于 SQL 语句来说，它们看似是普通的表，但在各个Worker节点之间进行水平分区。

也就是说，表的各个行存储在 table_1001、table_1002 等Worker表中。 Worker组件称为分片。

Citus 不仅在整个集群中运行 SQL语句，还运行 DDL 语句，因此更改分布式表的架构级联，进而更新跨Worker节点中的所有的表分片。

Distribution Column分布列

Citus 使用分片算法将行分配到分片。这意味着分配是确定性的——在我们的例子中基于表列（称为分布列）的值执行分配。集群管理员在分发表时必须指定该列。正如分布式数据建模的一般主题中所述，做出正确的选择对于性能和功能很重要。

类型 2：引用表

引用表是一种分布式表，其所有内容都集中到了单个分片中。该分片复制到了每一个Worker节点中。针对任何Worker节点的查询都可以在本地访问引用信息，而无需从另一个节点请求行，因此也不会产生此类网络开销。引用表没有分布列，因为无需区分每行的各个分片。

引用表通常很小，用于存储与在任何Worker节点上运行的查询相关的数据。例如，订单状态或产品类别等enumerated values。

当与引用表交互时，我们会自动对事务执行两阶段提交 (2PC)。这意味着无论是在写入、修改还是删除数据， Citus 将确保数据始终处于一致状态。

参考表部分将更详细的讨论这些表以及如何创建它们。

类型 3：本地表

当使用 Citus 时，连接并进行交互的Coordinator节点是安装了 Citus 扩展的常规 Spacture数据库。因此，可以在此Coordinator上创建普通表，并选择不将这些表分片。这对于不参与连接查询的小型管理表很有用。例如，用于执行应用程序登录和身份验证的用户表。

创建标准的 Spacture 表很容易，因为它是默认设置，是运行 CREATE TABLE 时得到的。在每个 Citus 部署中，我们几乎都会看到标准 Spacture 表与分布式表和引用表共存。事实上，如前所述，Citus本身就是使用本地表来保存集群元数据。

Shards

前面的部分将分片描述为在Worker节点内的较小表中包含分布式表的行的子集。本部分将更深入地介绍技术细节。

Coordinator上的 pg_dist_shard 元数据表包含系统中每个分布式表的每个分片的行。该行与分片 ID 相匹配，分片 ID 的范围是一组哈希整数 (shardminvalue, shardmaxvalue)。

SELECT * from  pg_dist_shard;

logicalrelid  | shardid | shardstorage | shardminvalue | shardmaxvalue

--------------+---------+--------------+---------------+---------------

github_events |  102026 | t            | 268435456     | 402653183

github_events |  102027 | t            | 402653184     | 536870911

github_events |  102028 | t            | 536870912     | 671088639

github_events |  102029 | t            | 671088640     | 805306367

(4 **rows**)

如果Coordinator节点要确定哪个分片包含 github_events 行，它将对该行中分布列的值执行哈希算法，并检查哪个分片的范围包含此哈希值。

分片放置（Shard Placements）

假设分片 102027 与相关行相关联。这意味着该行应该被读取或写入其中某个Worker节点中名为 github_events_102027 的表。是哪个Worker节点？这完全由元数据表决定，分片映射到Worker节点的过程称为分片放置。

加入一些元数据表给我们答案。这些是Coordinator为路由查询所做的查找类型。它将查询重写为引用特定表（如 github_events_102027）的片段，并在适当的Worker节点上运行这些片段。

SELECT
    shardid,
    node.nodename,
    node.nodeport
FROM pg_dist_placement placement
JOIN pg_dist_node node
  ON placement.groupid = node.groupid
 AND node.noderole = 'primary'::noderole
WHERE shardid = 102027;

┌─────────────┬─────────────┬─────────────┐
│   shardid   │  nodename   │   nodeport  │
├─────────────┼─────────────┼─────────────┤
│    102027   │  localhost  │     5433    │
└─────────────┴─────────────┴─────────────┘

在我们的 github_events 示例中，有四个分片。每个表的分片数量在整个集群中分布时都是可配置的。分片计数的最佳选择取决于的使用案例。

最后请注意，Citus 允许复制分片以防止数据丢失。有两种复制“模式”：Citus复制和流复制。前者创建额外的备份分片放置，并针对所有备份更新它们运行查询。后者效率更高，其利用Spacture的流式复制将每个节点的整个数据库备份到一个follower数据库。此数据库是透明的，不需要涉及 Citus 元数据表。

Co-Location

由于分片及其副本可以根据需求放置在节点上，因此将包含相关表的相关行的分片放在同一节点上是有意义的。这样，它们之间的连接查询可以避免通过网络发送更多信息，并且可以在单个 Citus 节点内执行。

一个例子是一个包含商店、产品和购买的数据库。如果所有三个表都包含并且由一个 store_id 列分布，那么限制到单个存储的所有查询都可以在单个Worker节点上高效运行。即使查询涉及这些表的任意组合，也是如此。

Parallelism

将查询分发在多台机器上允许一次运行更多查询，并允许通过向集群添加新机器来扩展处理速度。

将查询分布在多台机器上允许一次运行更多查询，并允许通过向集群添加新机器来扩展处理速度。此外，如上一节所述，将单个查询拆分为片段可以提高专门用于它的处理能力。后一种情况实现了最大的并行性，这意味着 CPU的利用率。

读取或影响均匀分布在许多节点上的分片的查询能够以“实时”速度运行。注意查询的结果还是需要通过Coordinator节点传回，所以当最终结果比较紧凑的时候加速最明显，比如计数和描述性统计等聚合函数。

查询处理解释了有关如何将查询分解为片段以及如何管理它们的执行的更多信息。

SQL参考文档

DDL

若要创建分布式表，需要首先定义表结构，可以使用 CREATE TABLE语句来定义表，就像对常规 Spacture 表执行此操作一样

CREATE TABLE github_events

(
    event_id bigint,
    event_type text,
    event_public boolean,
    repo_id bigint,
    payload jsonb,
    repo jsonb,
    actor jsonb,
    org jsonb,
    created_at timestamp
);

接下来，可以使用 create_distributed_table() 函数指定表分布列并在Worker节点上创建分片

SELECT create_distributed_table('github_events', 'repo_id');

create_distributed_table()函数将github_events表的数据基于repo_id列的值进行哈希分布。 create_distributed_table()函数还使用 citus.shard_count 和 citus.shard_replication_factor 配置值在Worker节点上创建分片。

它创建总数为 citus.shard_count 的分片，其中每个分片拥有部分哈希空间，并基于默认的 citus.shard_replication_factor 配置值进行复制。在Worker节点上创建的分片副本与Coordinator节点上的表具有相同的表结构引和约束定义。创建副本后，函数会将所有分布式元数据保存在Coordinator节点中。

为每个创建的分片分配一个唯一的分片 ID，其所有副本都具有相同的分片 ID。分片在Worker节点上表示为名为 'tablename_shardid' 的常规 Spacture表，其中 tablename 是分布式表的名称，而分片 ID 是分配的唯一 ID。可以连接到Worker postgres 实例，以查看或运行单个分片上的命令。

引用表

上述方法将表分发到多个水平分片中。另一种可能是将表分发到单个分片中，并将分片复制到每个Worker节点。以这种方式分发的表称为“引用表”。它们用于存储群集中的多个节点需要经常访问的数据。

引用表的常见候选项包括：

需要与较大的分布式表连接的小型表。
多租户应用中缺少租户 ID 列或不与租户关联的表。（在某些情况下，为了减少迁移工作，用户甚至可能会选择从与租户关联但当前缺少租户 ID 的表中创建引用表。）
需要跨多个列的唯一约束并足够小的表。

例如，假设多租户电子商务网站需要为其任何商店中的交易计算销售税。税务信息并不特定于任何租户。将其放在共享表中是有道理的。以US为中心的引用表可能如下所示：

-- a reference table

CREATE TABLE states (
  code char(2) PRIMARY KEY,
  full_name text NOT NULL,
  general_sales_tax numeric(4,3)
);


-- distribute it to all workers

SELECT create_reference_table('states');

现在，诸如计算购物车中税款的查询可以在states 表上连接，无网络开销，还可以将外键添加到状态代码中，以进行更好的验证。

除了将表作为单个复制的分片分发外,create_reference_table UDF 还将其标记为Citus元数据表中的引用表。 Citus自动执行两阶段提交 (2PC) 来修改以这种方式标记的表，这样可提供强大的一致性保证。

如果某个分布式表的分片计数为 1，则可以通过运行将其升级为可识别的引用表，如下所示：

SELECT upgrade_to_reference_table('table_name');

分发Coordinator节点数据

如果将现有的 Spacture 数据库实例转换为 Citus 集群的Coordinator节点，则其表中的数据可以高效地分发，并且最大程度地减少了应用程序的中断。

前面所述的 create_distributed_table 函数适用于空表和非空表，对于后者，它在整个群集中自动分发表行。将通过消息"NOTICE: Copying data from local table…"的显示来了解它是否执行此操作。例如：

CREATE TABLE series AS SELECT i
FROM generate_series(1,1000000) i;
SELECT create_distributed_table('series', 'i');
NOTICE:  Copying data from local able...

 create_distributed_table

 --------------------------

 (1 row)

在迁移数据时会阻止对表的写入，并且在函数提交后，待处理的写入将作为分布式查询进行处理。（如果该函数失败，则这些查询会再次成为本地查询。）查询可以继续正常进行，并且会在该函数提交后成为分布式查询。

在分配多个带有外键的表时，最好在运行 create_distributed_table 之前删除外键，并在分配表后重新创建外键。当一张表已分发而另一张表未分发时，不能总是强制执行外键。但是，分布式表和引用表之间支持外键。

将数据从外部数据库如从 Amazon RDS 迁移到 Citus Cloud时，首先通过 create_distributed_table 创建 Citus 分布式表，然后将数据复制到表中。

Co-Location

Co-Location将一组表的数据按照sharding规则，分配到相同的节点上，从而实现高效地关系操作，同时利用了整个数据集的水平可扩展性。

若要手动控制表的Co-Location组分配，请使用 create_distributed_table 的可选 colocate_with 参数。如果你不在意表的归置，则忽略此参数。它的默认值为 'default'，该值将表与具有相同分布列类型、分片计数和复制因子的任何其他默认归置表分组到一起。

-- these tables are implicitly co-located by using the same
-- distribution column type and shard count with the default
-- Co-Location group

SELECT create_distributed_table('A', 'some_int_col');
SELECT create_distributed_table('B', 'other_int_col');

如果新表与潜在的隐式Co-Location组中的其他表不相关，则指定 colocated_with⇒ 'none'。

-- not co-located with other tables

SELECT create_distributed_table('A', 'foo', colocate_with => 'none');

将不相关的表拆分为它们自己的Co-Location组将提高分片rebalance性能，因为同一组中的分片必须一起移动。

当表确实相关时（例如，当它们将被连接时），对它们进行显式Co-Location是有意义的。适当归置带来的收益比任何rebalance开销都重要。

若要对多个表进行显式归置，请先分发一张表，然后将其他表放入其Co-Location组。例如：

-- distribute stores

SELECT create_distributed_table('stores', 'store_id');

-- add to the same group as stores

SELECT create_distributed_table('orders', 'store_id', colocate_with =>
'stores');
SELECT create_distributed_table('products', 'store_id', colocate_with =>
'stores');

有关Co-Location组的信息存储在 pg_dist_colocation 表中，而 pg_dist_partition 显示哪些表分配给了哪些组。

删除表

可以使用标准的 Spacture DROP TABLE 命令删除分布式表。与常规表一样，DROP TABLE 删除目标表存在的任何索引、规则、触发器和约束。此外，它还删除Worker节点上的分片，并清除其元数据。

DROP TABLE github_events;

修改表

Citus 会自动分发类型的 DDL 语句，这意味着修改Coordinator节点上的分布式表也会更新Worker节点上的分片。部分 DDL 语句需要手动在Worker节点上执行，而某些部分 DDL 语句是被禁止的，例如会修改分布列的任何其他 DDL 语句。尝试运行不符合自动分发的 DDL 将引发错误并使Coordinator节点上的表保持不变。

下面是分发的 DDL 语句类别的参考。请注意，可以使用配置参数(citus.enable_ddl_propagation)启用或禁用自动分发。

添加/修改列

Citus 会自动分发大多数 ALTER TABLE 命令。添加列或更改其默认值的工作方式与在单机 Spacture 数据库中的工作方式相同：

-- Adding a column
ALTER TABLE products ADD COLUMN description text;

-- Changing default value
ALTER TABLE products ALTER COLUMN price
SET DEFAULT 7.77;

也可以对现有列进行更改（例如，对其进行重命名或更改其数据类型）。但是无法更改分布列的数据类型。此列确定表数据在 Citus 集群中的分布方式，并且修改其数据类型将需要移动数据。

如果尝试这样做，将会导致错误：

-- assumining store_id is the distribution column
-- for products, and that it has type integer

ALTER TABLE products
ALTER COLUMN store_id TYPE text;

/*

ERROR:  XX000: cannot execute ALTER TABLE command involving partition
column
LOCATION:  ErrorIfUnsupportedAlterTableStmt, multi_utility.c:2150

*/

添加/删除约束

使用 Citus 可以继续享受关系数据库的安全性，包括数据库约束。由于分布式系统的性质，Citus 不会在Worker节点之间交叉引用唯一性约束或引用完整性。

在这些情况下可能会创建外键：

在两个本地（非分布式）表之间；
当键包含分布列时，在两个Co-Location分布式表之间；
作为引用参考表的分布式表

不支持引用表作为外键约束的引用表，即不支持从引用到引用和对分布式引用的键。

要在Co-Location分布式表之间设置外键，请始终在键中包含分布列。

Note: 主键和唯一性约束必须包括分布列。将它们添加到非分布列将产生错误

此示例显示如何在分布式表中创建主键和外键：

--

-- Adding a primary key

-- --------------------

-- We'll distribute these tables on the account_id. The ads and clicks
-- tables must use compound keys that include account_id.__

ALTER TABLE accounts ADD PRIMARY KEY (id);
ALTER TABLE ads ADD PRIMARY KEY (account_id, id);
ALTER TABLE clicks ADD PRIMARY KEY (account_id, id);

-- Next distribute the tables

SELECT create_distributed_table('accounts','id');
SELECT create_distributed_table('ads',      'account_id');
SELECT create_distributed_table('clicks',  'account_id');

--

-- Adding foreign keys

-- -------------------

-- Note that this can happen before or after distribution, as long
as
-- there exists a uniqueness constraint on the target column(s)
which
-- can only be enforced before distribution.

ALTER TABLE ads ADD CONSTRAINT ads_account_fk
  FOREIGN KEY (account_id) REFERENCES accounts (id);
ALTER TABLE clicks ADD CONSTRAINT clicks_ad_fk
  FOREIGN KEY (account_id, ad_id) REFERENCES ads (account_id, id);

同样，在唯一性约束中包含分布列：

-- Suppose we want every ad to use a unique image. Notice we can
-- enforce it only per account when we distribute by account id.

ALTER TABLE ads ADD CONSTRAINT ads_unique_image
  UNIQUE (account_id, image_url);

Not-null 约束可以应用于任何列（分布式或非分布式），因为它们不需要在Worker节点之间进行查找。

ALTER TABLE ads ALTER COLUMN image_url SET NOT NULL;

使用 NOT VALID 约束

在某些情况下，对新行强制实施约束可能会很有用，同时允许现有的非一致性行保持不变。Citus使用 Spacture 的 "NOT VALID" 约束名称，支持 CHECK 约束和外键的此功能。

例如，考虑将用户配置文件存储在引用表中的应用程序。

-- we're using the "text" column type here, but a real application
-- might use "citext" which is available in a postgres contrib
module

CREATE TABLE users ( email text PRIMARY KEY );
SELECT create_reference_table('users');

最后，假设有几个non-address进入表中。

INSERT INTO users VALUES

   ('foo@example.com'), ('hacker12@aol.com'), ('lol');

我们想要验证地址，但 Spacture 通常不允许添加对现有行失败的 CHECK 约束。但是，Spacture允许将约束标记为无效：

ALTER TABLE users
ADD CONSTRAINT syntactic_email
CHECK (email ~

'^[a-zA-Z0-9.!#$%&''*+/=?^_`\{|}~-]+@[a-zA-Z0-9](?:[a-zA-Z0-9-]\{0,61}[a-zA-Z0-9])?(?:\.[a-zA-Z0-9](?:[a-zA-Z0-9-]\{0,61}[a-zA-Z0-9])?)*$'
) NOT VALID;

新的行现在受保护。

INSERT INTO users VALUES ('fake');

/*

ERROR:  new row for relation "users_102010" violates
        check constraint "syntactic_email_102010"
DETAIL:  Failing row contains (fake).

*/

稍后，在非高峰时段，数据库管理员可以尝试修复错误的行并重新验证约束。

-- later, attempt to validate all rows
ALTER TABLE users
VALIDATE CONSTRAINT syntactic_email;

添加/删除索引

Citus 支持添加和删除索引：

-- Adding an index

CREATE INDEX clicked_at_idx ON clicks USING BRIN
(clicked_at);

-- Removing an index

DROP INDEX clicked_at_idx;

添加索引需要使用写锁，这在多租户"记录系统"中可能并不合适。为了最大限度地减少应用程序停机时间，请改为并发创建索引。此方法需要的总工作量超过标准索引构建的时间，并且需要更长的时间才能完成。但是，因为它允许在构建索引时继续正常操作，因此此方法对于在生产环境中添加新索引很有用。

-- Adding an index without locking table writes

CREATE INDEX CONCURRENTLY clicked_at_idx ON clicks USING BRIN
(clicked_at);

手动修改

目前其他 DDL 命令不会自动分发，但是可以手动分发更改。

DML

要将数据插入分布式表，可以使用标准的 Spacture INSERT 命令。例如，我们从 Github Archive 数据集中随机选取两行。

/*

CREATE TABLE github_events

(
  event_id bigint,
  event_type text,
  event_public boolean,
  repo_id bigint,
  payload jsonb,
  repo jsonb,
  actor jsonb,
  org jsonb,
  created_at timestamp
);

*/


INSERT INTO github_events VALUES
(2489373118,'PublicEvent','t',24509048,'\{}','\{"id": 24509048, "url":
"https://api.github.com/repos/SabinaS/csee6868", "name":
"SabinaS/csee6868"}','\{"id": 2955009, "url":
"https://api.github.com/users/SabinaS", "login": "SabinaS",
"avatar_url": "https://avatars.githubusercontent.com/u/2955009?",
"gravatar_id": ""}',NULL,'2015-01-01 00:09:13');

INSERT INTO github_events VALUES
(2489368389,'WatchEvent','t',28229924,'\{"action": "started"}','\{"id":
28229924, "url": "https://api.github.com/repos/inf0rmer/blanket",
"name": "inf0rmer/blanket"}','\{"id": 1405427, "url":
"https://api.github.com/users/tategakibunko", "login": "tategakibunko",
"avatar_url": "https://avatars.githubusercontent.com/u/1405427?",
"gravatar_id": ""}',NULL,'2015-01-01 00:00:24');

向分布式表中插入行时，必须指定被插入行的分布列。根据分布列，Citus 确定插入的数据应路由到的正确分片。然后，将查询转发到正确的分片，并在该分片的所有副本上执行插入命令。

有时将多个 insert 语句放在一起成为多行的单个 insert 很方便。它也可以比重复进行数据库查询更有效。例如，上一节中的示例可以像这样一次性加载：

INSERT INTO github_events VALUES

 (

2489373118,'PublicEvent','t',24509048,'\{}','\{"id": 24509048, "url":
"https://api.github.com/repos/SabinaS/csee6868", "name":
"SabinaS/csee6868"}','\{"id": 2955009, "url":
"https://api.github.com/users/SabinaS", "login": "SabinaS",
"avatar_url": "https://avatars.githubusercontent.com/u/2955009?",
"gravatar_id": ""}',NULL,'2015-01-01 00:09:13'

  ), (

 2489368389,'WatchEvent','t',28229924,'\{"action": "started"}','\{"id":
28229924, "url": "https://api.github.com/repos/inf0rmer/blanket",
"name": "inf0rmer/blanket"}','\{"id": 1405427, "url":
"https://api.github.com/users/tategakibunko", "login": "tategakibunko",
"avatar_url": "https://avatars.githubusercontent.com/u/1405427?",
"gravatar_id": ""}',NULL,'2015-01-01 00:00:24'

  );

“From Select”子句（分布式Rollup）

Citus 还支持 INSERT … SELECT 语句——根据选择查询的结果插入行。这是一种数据导入的便捷方法，并且还允许使用 ON CONFLICT 子句进行“更新插入”，这是进行分布式Rollup的最简单方法。

在 Citus 中有两种方法可以从 select 语句中插入。第一个是源表和目标表是否Co-Location，并且选择/插入语句都包含分布列。在这种情况下，Citus 可以下推 INSERT … SELECT 语句以在所有节点上并行执行。

执行 INSERT … SELECT 语句的第二种方式是从Worker节点选择结果，将数据拉到Coordinator节点，然后从Coordinator节点发出带有数据的 INSERT 语句。当源表和目标表不在同一位置时，Citus 被迫使用这种方法。由于网络开销，这种方法效率不高。

如果 upsert 是应用程序中的重要操作，理想的解决方案是对数据建模，以便源表和目标表位于同一位置，并且，如果聚合分布列可以成为 upsert 语句中 GROUP BY 子句的一部分。这允许操作跨Worker节点并行运行以获得最大速度。

如果对 Citus 使用的方法有疑问，请使用 EXPLAIN 命令，如 Spacture调优中所述。

COPY 命令（批量加载）

要从文件中批量加载数据，可以直接使用 Spacture 的 \COPY 命令。

首先通过运行下载我们的示例 github_events 数据集：

wget
http://examples.citusdata.com/github_archive/github_events-2015-01-01-\{0..5}.csv.gz
gzip -d github_events-2015-01-01-*.gz

然后，可以使用 psql 复制数据：

\COPY github_events FROM 'github_events-2015-01-01-0.csv' WITH (format
CSV)

Note: 没有跨分片快照隔离的概念，这意味着与 COPY 并发运行的多分片 SELECT 可能会看到它在某些分片上提交，但不会在其他分片上提交。如果用户正在存储事件数据，他可能偶尔会观察到最近数据中的小差距。如果这是一个问题，则由应用程序来处理（例如，从查询中排除最近的数据，或使用一些锁）。

如果执行COPY命令时，无法和shard所在的节点建立连接，那么它的行为方式与 INSERT 相同，即将shard placement标记为非活动，除非没有更多可用的shard placement。如果连接后发生任何其他故障，事务将回滚，因此不会更改元数据。

使用汇总缓存聚合

事件数据管道和实时仪表板等应用程序需要对大量数据进行亚秒级查询。使这些查询快速的一种方法是提前计算和保存聚合。这称为“rolling up”数据，它避免了在运行时处理原始数据的成本。作为额外的好处，将时间序列数据汇总为每小时或每天的统计数据还可以节省空间。当不再需要旧数据的全部细节并且聚合足够时，可能会删除旧数据。

例如，这是一个用于按 url 跟踪页面浏览量的分布式表：

CREATE TABLE page_views (
  site_id int,
  url text,
  host_ip inet,
  view_time timestamp default now(),
  PRIMARY KEY (site_id, url)
);

SELECT create_distributed_table('page_views', 'site_id');

一旦表中填充了数据，我们就可以运行aggregate查询来计算每天每个 URL 的页面浏览量，限制为给定的站点和年份。

-- how many views per url per day on site 5?

SELECT view_time::date AS day, site_id, url, count(*) AS view_count
  FROM page_views
  WHERE site_id = 5 AND
    view_time >= date '2016-01-01' AND view_time < date '2017-01-01'
  GROUP BY view_time::date, site_id, url;

上述设置有效，但有两个缺点。首先，当重复执行aggregate查询时，它必须遍历每个相关行并重新计算整个数据集的结果。如果使用此查询来呈现仪表板，则将汇总结果保存在每日页面视图表中并查询该表会更快。其次，存储成本将随着数据量和可查询历史的长度成比例增长。在实践中，可能希望在短时间内保留原始事件并查看较长时间窗口内的历史图表。

为了获得这些好处，我们可以创建一个 daily_page_views 表来存储每日统计信息。

CREATE TABLE daily_page_views (
  site_id int,
  day date,
  url text,
  view_count bigint,
  PRIMARY KEY (site_id, day, url)
);

SELECT create_distributed_table('daily_page_views', 'site_id');

在这个例子中，我们在 site_id 列上分布了 page_views 和 daily_page_views。这确保与特定站点对应的数据将共同位于同一节点上。在每个节点上将两个表的行保持在一起可以最大限度地减少节点之间的网络传输并实现高度并行执行。

一旦我们创建了这个新的分布式表，我们就可以运行 INSERT INTO … SELECT 将原始页面视图汇总到聚合表中。在下文中，我们汇总了每天的页面浏览量。 Citus 用户通常在一天结束后等待一段时间来运行这样的查询，以适应延迟的数据。

-- roll up yesterday's data
INSERT INTO daily_page_views (day, site_id, url, view_count)
  SELECT view_time::date AS day, site_id, url, count(*) AS view_count
  FROM page_views
  WHERE view_time >= date '2017-01-01' AND view_time < date '2017-01-02'
  GROUP BY view_time::date, site_id, url;

__-- now the results are available right out of the table__
SELECT day, site_id, url, view_count
  FROM daily_page_views
  WHERE site_id = 5 AND
    day >= date '2016-01-01' AND day < date '2017-01-01';

上面的汇总查询聚合了前一天的数据并将其插入到 daily_page_views 中。每天运行一次查询意味着不需要更新汇总表行，因为新一天的数据不会影响以前的行。

在处理延迟的数据或每天多次运行汇总查询时，情况会发生变化。如果任何新行与汇总表中已有的天数匹配，则匹配计数应增加。 Spacture 可以使用“ON CONFLICT”来处理这种情况，这是它执行更新插入的技术。这是一个例子。

-- roll up from a given date onward,
-- updating daily page views when necessary
INSERT INTO daily_page_views (day, site_id, url, view_count)
  SELECT view_time::date AS day, site_id, url, count(*) AS view_count
  FROM page_views
  WHERE view_time >= date '2017-01-01'
  GROUP BY view_time::date, site_id, url
  ON CONFLICT (day, url, site_id) DO UPDATE SET
    view_count = daily_page_views.view_count + EXCLUDED.view_count;

更新和删除

可以使用标准的 Spacture UPDATE 和 DELETE 命令更新或删除分布式表中的行。

DELETE FROM github_events
WHERE repo_id IN (24509048, 24509049);

UPDATE github_events
SET event_public = TRUE
WHERE (org->>'id')::int = 5430905;

当更新/删除影响多个分片时，如上例所示，Citus 默认使用单阶段提交协议。为了提高安全性，可以通过设置启用两阶段提交

SET citus.multi_shard_commit_protocol = '2pc';

如果更新或删除仅影响单个分片，则它在单个Worker节点内运行。在这种情况下，不需要启用 2PC。当按表的分布列更新或删除过滤时，通常会发生这种情况：

-- since github_events is distributed by repo_id,
-- this will execute in a single worker node

DELETE FROM github_events
WHERE repo_id = 206084;

此外，在处理单个分片时，Citus 支持 SELECT … FOR UPDATE。这是对象关系映射器 (ORM) 有时使用的一种技术，用于安全地：

加载行
在应用程序代码中进行计算
根据计算更新行

选择要更新的行会对它们施加写锁，以防止其他进程导致“Lost-update”异常。

BEGIN;

  -- select events for a repo, but
  -- lock them for writing

  SELECT *
  FROM github_events
  WHERE repo_id = 206084
  FOR UPDATE;

  -- calculate a desired value event_public using
  -- application logic that uses those rows...

  -- now make the update
  UPDATE github_events
  SET event_public = :our_new_value
  WHERE repo_id = 206084;

COMMIT;

此功能仅支持散列分布式表和引用表，并且仅支持 replication_factor 为 1 的那些。

最大化写入性能

INSERT 和 UPDATE/DELETE 语句都可以在大型机器上扩展到每秒大约 50,000 个查询。但是，要达到这个速度，将需要使用许多并行的、长期存在的连接并考虑如何处理锁定。

查询分布式表（SQL）

正如前几节所讨论的，Citus 是一个扩展，它扩展了最新的 Spacture以实现分布式执行。这意味着可以在 Citus Coordinator上使用标准的Spacture SELECT 查询进行查询。然后，Citus将分布式执行涉及复杂选择、分组和排序以及 JOIN 的 SELECT查询，以加快查询性能。在高层次上，Citus 将 SELECT 查询划分为更小的查询片段，将这些查询片段分配给Worker节点，监控他们的执行，合并他们的结果（并在需要时对其进行排序），并将最终结果返回给用户。

在以下部分中，我们将讨论使用 Citus 运行的不同类型的查询。

聚合函数（Aggregate Functions）

Citus 支持并分布式执行 Spacture 支持的大多数聚合函数。 Citus 的查询planner将聚合转换为其可交换和关联形式，以便可以分布式执行。在此过程中，Worker节点对分片运行聚合查询，然后Coordinator节点将Worker节点的结果组合起来以产生最终输出。

Count(Distinct)聚合

Citus 以多种方式支持Count(Distinct)聚合。如果Count(Distinct)聚合在分布列上，Citus 可以直接将查询下推给Worker节点。如果没有，Citus 对每个Worker节点运行 select distinct 语句，并将列表返回给Coordinator节点，在那里它获得最终结果。

请注意，当Worker节点拥有多个distinct列时，传输此数据会变慢。对于包含多个Coun(Distinct)聚合的查询尤其如此，例如：

-- multiple distinct counts in one query tend to be slow
SELECT count(distinct a), count(distinct b), count(distinct c)
FROM table_abc;

对于这些类型的查询，Worker节点上执行的查询基本上会产生要传输到Coordinator节点的行的cross-product 。

Limit下推

Citus 还尽可能将Limit子句下推到Worker节点的分片上，以最大限度地减少网络传输的数据量。

但是，在某些情况下，带有 LIMIT 子句的 SELECT 查询可能需要从每个分片中获取所有行以生成正确的结果。例如，如果查询需要按聚合列排序，则需要所有分片中该列的结果来确定最终聚合值。由于大量网络数据传输，这会降低 LIMIT 子句的性能。在这种情况下，如果近似会产生有意义的结果，Citus 为近似 LIMIT 子句提供了一个选项。

LIMIT 近似默认是禁用的，可以通过设置配置参数 citus.limit_clause_row_fetch_count 来启用。在这个配置值的基础上，Citus 会限制每个任务返回的行数，以便在Coordinator节点上聚合。由于此限制，最终结果可能是近似的。增加此限制将提高最终结果的正确性，同时仍然提供从Worker节点中提取的行数的上限。

SET citus.limit_clause_row_fetch_count to 10000;

分布式表视图 (Views on Distributed Tables)

Citus 支持分布式表上的所有视图。

请注意，某些视图会导致查询计划的效率低于其他视图。

Citus 也支持物化视图，并将它们作为本地表存储在Coordinator节点上。在分布式查询中使用物化视图需要将它们包装在子查询中。

Joins

Citus 支持任意数量的表之间的 equi-JOIN，而不管它们的大小和分布方法如何。查询planner根据表的分布方式选择最佳join方法和join顺序。它评估几种可能的连接顺序并创建一个Join查询计划，该计划依赖通过网络传输的最少数据。

Co-located Joins

当两个表是Co-Location时，它们可以有效地连接到它们的公共分布列上。 Co-Location Join是连接两个大型分布式表的最有效方式。

在内部，Citus Coordinator节点通过查看分布列元数据知道Co-Location表的哪些分片可能与另一个表的分片匹配。这允许 Citus 删除不能产生匹配连接键的分片对。剩余分片对之间的连接在Worker节点上并行执行，然后将结果返回给Coordinator节点。

Note: 确保表分布到相同数量的分片中，并且每个表的分布列具有完全匹配的类型。尝试连接类型不同的列（如 int 和 bigint）可能会导致问题。

引用表连接 (Reference table joins)

参考表可用作“维度”表，以有效连接大型“事实”表。因为引用表在所有Worker节点上完全复制，所以引用连接可以分解为每个Worker节点上的本地连接并并行执行。引用连接类似于Co-Location join的更灵活版本，因为引用表不分布在任何特定列上，并且可以自由地连接到它们的任何列上

Repartition joins

在某些情况下，可能需要在分布列以外的列上连接两个表。对于这种情况，Citus 还允许通过为查询动态重新分区表来连接非分布键列。

在这种情况下，要分区的表由查询优化器根据分布列、连接键和表的大小确定。使用重新分区的表，可以确保只有相关的分片对相互连接，从而大大减少了跨网络传输的数据量。

通常，Co-Location join比repartition join更有效，因为repartition join需要对数据进行Shuffle。因此，应该尽可能尝试通过公共连接键来分布表。

查询处理

Citus 集群由一个Coordinator节点和多个Worker节点组成。数据在Worker节点上分片和复制，而Coordinator节点存储有关这些分片的元数据。向集群发出的所有查询都通过Coordinator节点执行。 Coordinator节点将查询划分为更小的查询片段，其中每个查询片段可以在分片上独立运行。然后Coordinator节点将查询片段分配给Worker节点，监控他们的执行，合并他们的结果，并将最终结果返回给用户。查询处理架构可以通过下图来简要描述。

Citus 的查询处理管道涉及两个组件：

分布式查询planner和executor
Spacture planner和executor

*分布式查询planner *

Citus 的分布式查询planner接收 SQL 查询，并生成分布式查询计划进行分布式执行。

对于 SELECT 查询，planner首先创建输入查询的计划树并将其转换为其可交换和关联形式，以便可以并行化。它还应用了多项优化，以确保以可以线性扩展方式执行查询，并最大限度地减少网络 I/O。

接下来，planner将查询分为两部分 - 在Coordinator节点上运行的Coordinator节点需要执行的查询和在Worker节点上的各个分片上运行的查询片段。然后planner将这些查询片段分配给Worker节点，以便有效地使用他们的所有资源。在这一步之后，分布式查询计划被传递给分布式executor执行。

分布列上的键值查找或DML查询的生成查询计划过程略有不同，因为它们恰好命中一个分片。一旦planner收到传入的查询，它需要决定查询应该路由到的正确分片。为此，它提取传入行中的分布列并查找元数据以确定查询的正确分片。然后，planner重写该命令的 SQL 以引用分片表而不是原始表。重写的planner被传递给分布式executor。

分布式查询executor

Citus 的分布式executor运行分布式planner并处理查询执行期间发生的故障。执行者连接到Worker节点，将分配的任务发送给他们并监控他们的执行。如果 executor 无法将任务分配给指定的Worker节点，或者任务执行失败，则 executor 会动态地将任务重新分配给其他 Worker 上的副本。executor在处理失败时只处理失败的子查询，而不是整个查询。

Citus 有两种executor类型：adaptive 和task-tracker。它根据每个查询的结构动态选择使用哪个，并且可以对单个查询同时使用多个，根据需要将不同的executor分配给不同的子查询/CTE 以支持 SQL 功能。此过程是递归的：如果 Citus 无法确定如何运行子查询，则它会检查嵌套的子查询。

在高层次上，adaptive executor用于大多数查询。但是，repartition_join 需要task-tracker。可以通过运行 Spacture 的 EXPLAIN 命令来显示每个查询的executor选择。这对于调试性能问题很有用。

Adaptive Executor

adaptive executor是默认的。它非常适合快速响应涉及过滤、聚合和Co-Location Join的查询，以及运行具有完整 SQL 覆盖范围的单租户查询。adaptive executor根据需要为每个分片打开一个连接，并将所有片段查询发送给他们。然后它从每个片段查询中获取结果，合并它们，并将最终结果返回给用户。

此外，当adaptive executor检测到简单的 INSERT、UPDATE 或 DELETE 查询时，它会将传入的查询分配给具有目标分片的Worker节点。然后由对应的Worker节点处理查询，并将结果返回给用户。如果分片副本上的修改失败，executor会将相应的分片副本标记为无效，以保持数据一致性。

Task Tracker Executor

task-tracker executor非常适合长时间运行的复杂数据仓库查询。这个executor只为每个Worker节点打开一个连接，并将所有片段查询分配给Worker节点上的task-tracker守护进程。task-tracker守护进程然后定期安排新的task, 并查看它们的完成情况。Coordinator节点上的executor定期检查这些task-tracker，看看它们的task是否完成。

Worker上的每个task-tracker守护进程也确保同时执行 citus.max_running_tasks_per_node。当查询不是从内存提供时，此并发限制有助于避免磁盘 I/O 争用。task-tracker旨在有效处理复杂的查询，这些查询需要在Worker节点之间重新分区和shuffle中间数据。

子查询/CTE 推拉（push-pull）执行

如有必要，Citus 可以将来自子查询和 CTE 的结果汇集到Coordinator节点中，然后将它们推送回Worker节点以供外部查询使用。这允许 Citus 支持更多类型的 SQL，甚至可以在查询及其子查询之间混合executor类型。

例如，在 WHERE 子句中包含子查询有时无法与主查询同时内联执行，而必须单独执行。假设一个 Web 分析应用程序维护一个按 page_id 分区的访问表。要查询前 20 个访问量最大的页面上的访问者会话数，我们可以使用子查询来查找页面列表，然后使用外部查询来计算会话数。

SELECT page_id, count(distinct session_id)
FROM visits
WHERE page_id IN (
  SELECT page_id
  FROM visits
  GROUP BY page_id
  ORDER BY count(*) DESC
  LIMIT 20
)
GROUP BY page_id;

实时executor希望通过 page_id 对每个分片运行此查询的片段，计算不同的 session_id，并在Coordinator节点上汇集结果。但是子查询中的 LIMIT 意味着子查询不能作为片段的一部分执行。通过‘递归计划 (recursively planning) ’查询，Citus 可以单独运行子查询，将结果推送给所有Worker节点，运行主片段查询，并将结果拉回Coordinator节点。 “push-pull”设计支持像上面那样的子查询。

让我们通过查看此查询的 EXPLAIN 输出来了解这一点。它相当涉及：

GroupAggregate  (cost=0.00..0.00 rows=0 width=0)
  Group Key: remote_scan.page_id
  ->  Sort  (cost=0.00..0.00 rows=0 width=0)
    Sort Key: remote_scan.page_id
    ->  Custom Scan (Citus Real-Time)  (cost=0.00..0.00 rows=0 width=0)
      ->  Distributed Subplan 6_1
        ->  Limit  (cost=0.00..0.00 rows=0 width=0)
          ->  Sort  (cost=0.00..0.00 rows=0 width=0)
            Sort Key:
COALESCE((pg_catalog.sum((COALESCE((pg_catalog.sum(remote_scan.worker_column_2))::bigint,
'0'::bigint))))::bigint, '0'::bigint) DESC
            ->  HashAggregate  (cost=0.00..0.00 rows=0 width=0)
              Group Key: remote_scan.page_id
              ->  Custom Scan (Citus Real-Time)  (cost=0.00..0.00 rows=0
width=0)
                Task Count: 32
                Tasks Shown: One of 32
                ->  Task
                  Node: host=localhost port=5433 dbname=postgres
                  ->  Limit  (cost=1883.00..1883.05 rows=20 width=12)
                    ->  Sort  (cost=1883.00..1965.54 rows=33017 width=12)
                      Sort Key: (count(*)) DESC
                      ->  HashAggregate  (cost=674.25..1004.42 rows=33017 width=12)
                        Group Key: page_id
                        ->  Seq Scan on visits_102264 visits
                        (cost=0.00..509.17 rows=33017 width=4)
      Task Count: 32
      Tasks Shown: One of 32
      ->  Task
        Node: host=localhost port=5433 dbname=postgres
        ->  HashAggregate  (cost=734.53..899.61 rows=16508 width=8)
          Group Key: visits.page_id, visits.session_id
          ->  Hash Join  (cost=17.00..651.99 rows=16508 width=8)
            Hash Cond: (visits.page_id = intermediate_result.page_id)
            ->  Seq Scan on visits_102264 visits  (cost=0.00..509.17 rows=33017 width=8)
            ->  Hash  (cost=14.50..14.50 rows=200 width=4)
              ->  HashAggregate  (cost=12.50..14.50 rows=200 width=4)
                Group Key: intermediate_result.page_id
                ->  Function Scan on read_intermediate_result intermediate_result  (cost=0.00..10.00 rows=1000 width=4)

让我们把它拆开并检查每一块。

GroupAggregate  (cost=0.00..0.00 rows=0 width=0)
  Group Key: remote_scan.page_id
  ->  Sort  (cost=0.00..0.00 rows=0 width=0)
    Sort Key: remote_scan.page_id

查询计划的根节点是Coordinator节点对各个Worker节点的结果所做的事情。在这种情况下，它对它们进行分组，而 GroupAggregate 要求首先对它们进行排序。

->  Custom Scan (Citus Real-Time)  (cost=0.00..0.00 rows=0 width=0)
      ->  Distributed Subplan 6_1

自定义扫描有两个大的子节点，从“分布式子计划”开始。

->  Limit  (cost=0.00..0.00 rows=0 width=0)
          ->  Sort  (cost=0.00..0.00 rows=0 width=0)
            Sort Key:
COALESCE((pg_catalog.sum((COALESCE((pg_catalog.sum(remote_scan.worker_column_2))::bigint,
'0'::bigint))))::bigint, '0'::bigint) DESC
            ->  HashAggregate  (cost=0.00..0.00 rows=0 width=0)
              Group Key: remote_scan.page_id
              ->  Custom Scan (Citus Real-Time)  (cost=0.00..0.00 rows=0 width=0)
                Task Count: 32
                Tasks Shown: One of 32
                ->  Task
                  Node: host=localhost port=5433 dbname=postgres
                  ->  Limit  (cost=1883.00..1883.05 rows=20 width=12)
                    ->  Sort  (cost=1883.00..1965.54 rows=33017 width=12)
                      Sort Key: (count(*)) DESC
                      ->  HashAggregate  (cost=674.25..1004.42  rows=33017 width=12)
                        Group Key: page_id
                        ->  Seq Scan on visits_102264 visits (cost=0.00..509.17 rows=33017 width=4)

Worker节点为 32 个分片中的每一个运行上述内容（Citus 选择一个代表进行显示）。我们可以识别 IN (…) 子查询的所有部分：排序、分组和限制。当所有Worker节点完成此查询后，他们将输出发送回Coordinator节点，Coordinator节点将其作为“中间结果”放在一起。

Task Count: 32
      Tasks Shown: One of 32
      ->  Task
        Node: host=localhost port=5433 dbname=postgres
        ->  HashAggregate  (cost=734.53..899.61 rows=16508 width=8)
          Group Key: visits.page_id, visits.session_id
          ->  Hash Join  (cost=17.00..651.99 rows=16508 width=8)
            Hash Cond: (visits.page_id = intermediate_result.page_id)

Citus 在第二个子树中启动另一个实时作业。它将计算访问中的不同会话。它使用 JOIN 连接中间结果。中间结果将帮助它限制在前二十页。

->  Seq Scan on visits_102264 visits  (cost=0.00..509.17 rows=33017
width=8)
            ->  Hash  (cost=14.50..14.50 rows=200 width=4)
              ->  HashAggregate  (cost=12.50..14.50 rows=200 width=4)
                Group Key: intermediate_result.page_id
                ->  Function Scan on read_intermediate_result intermediate_result  (cost=0.00..10.00 rows=1000 width=4)

Worker节点使用 read_intermediate_result 函数在内部检索中间结果，该函数从从Coordinator节点复制的文件中加载数据。

此示例展示了 Citus 如何使用分布式子计划在多个步骤中执行查询，以及如何使用 EXPLAIN 了解分布式查询执行。

Spacture planner和executor (Spacture planner and executor)

一旦分布式executor将查询片段发送给Worker节点，它们就会像常规 Spacture 查询一样被处理。该Worker节点上的Spacture planner选择最佳计划以在相应的分片表上本地执行该查询。 Spacture 执行程序然后运行该查询并将查询结果返回给分布式executor。最后，分布式executor将结果传递给Coordinator节点进行最终聚合。

手动查询传播

当用户发出查询时，Citus Coordinator节点将其划分为较小的查询片段，其中每个查询片段可以在分片上独立运行。这允许 Citus 跨节点分发每个查询。

然而，查询划分为片段的方式（以及分发哪些查询）因查询类型而异。在某些高级情况下，手动控制此行为很有用。 Citus 提供了实用函数来将 SQL 分发到Worker节点。

手动查询分发绕过Coordinator节点、锁定和任何其他一致性检查。这些函数可用作允许 Citus执行暂时不支持的查询的最后手段。小心使用它们以避免数据不一致和死锁。

在所有Worker节点上运行

最细粒度的执行级别是广播一条语句以在所有Worker节点上执行。这对于查看整个数据库的属性或在整个集群中统一创建 UDF 非常有用。例如：

-- Make a UDF available on all workers
SELECT run_command_on_workers($cmd$ CREATE FUNCTION ... $cmd$);

-- List the work_mem setting of each worker database
SELECT run_command_on_workers($cmd$ SHOW work_mem; $cmd$);

本节中的 run_command_on_workers 函数和其他手动分发命令只能运行返回单列单行的查询。

在所有分片上运行

下一个粒度级别是在特定分布式表的所有分片上运行命令。例如，它可以用于直接在Worker节点上读取表的属性。在Worker节点本地运行的查询可以完全访问元数据，例如表统计信息。

run_command_on_shards 函数将 SQL 命令应用于每个分片，其中分片名称用于在命令中进行动态替换。这是通过使用每个Worker节点上的 pg_class 表来估计每个分片的行数来估计分布式表的行数的示例。请注意 %s 将被替换为每个分片的名称。

-- Get the estimated row count for a distributed table by summing
the
-- estimated counts of rows for each shard.
SELECT sum(result::bigint) AS estimated_count
  FROM run_command_on_shards(
    'my_distributed_table',
    $cmd$
      SELECT reltuples
        FROM pg_class c
        JOIN pg_catalog.pg_namespace n on n.oid=c.relnamespace
       WHERE (n.nspname || '.' || relname)::regclass = '%s'::regclass

         AND n.nspname NOT IN ('citus', 'pg_toast', 'pg_catalog')
    $cmd$
  );

在所有Shard placement上运行

最精细的执行级别是在所有分片及其副本（也称为shard placement）上运行命令。它可用于运行数据修改命令，这些命令必须应用于每个副本以确保一致性。

例如，假设一个分布式表有一个 updated_at 字段，我们想要“修改”所有行，以便在某个时间将它们标记为已更新。通过Coordinator节点执行的普通 UPDATE 语句需要通过分布列进行过滤，但我们可以在所有分片和副本之间手动分发更新：

-- note we're using a hard-coded date rather than
-- a function such as "now()" because the query will
-- run at slightly different times on each replica

SELECT run_command_on_placements(
  'my_distributed_table',
  $cmd$
    UPDATE %s SET updated_at = '2017-01-01';
  $cmd$
);

run_command_on_placements 的一个有用的辅助函数是 run_command_on_colocated_placements。它将位于同一位置的分布式表的两个Shard分配的名称插入到查询中。放置对始终被选择为位于完全 SQL 覆盖可用的同一Worker节点的本地。因此，我们可以使用触发器等高级 SQL 功能来关联表：

-- Suppose we have two distributed tables
CREATE TABLE little_vals (key int, val int);
CREATE TABLE big_vals    (key int, val int);
SELECT create_distributed_table('little_vals', 'key');
SELECT create_distributed_table('big_vals',    'key');

-- We want to synchronise them so that every time little_vals

-- are created, big_vals appear with double the value
--
-- First we make a trigger function on each worker, which will
-- take the destination table placement as an argument
SELECT run_command_on_workers($cmd$
  CREATE OR REPLACE FUNCTION embiggen() RETURNS TRIGGER AS $$
    BEGIN
      IF (TG_OP = 'INSERT') THEN
        EXECUTE format(
          'INSERT INTO %s (key, val) SELECT ($1).key, ($1).val*2;',
          TG_ARGV[0]
        ) USING NEW;
      END IF;
      RETURN NULL;
    END;
  $$ LANGUAGE plpgsql;
$cmd$);
-- Next we relate the co-located tables by the trigger function
-- on each co-located placement
SELECT run_command_on_colocated_placements(
  'little_vals',
  'big_vals',
  $cmd$
    CREATE TRIGGER after_insert AFTER INSERT ON %s
      FOR EACH ROW EXECUTE PROCEDURE embiggen(%s)
  $cmd$
);

限制

对于多语句事务，没有防止死锁的安全措施。
没有针对中间查询失败和由此产生的不一致的安全措施。
查询结果缓存在内存中；这些函数无法处理非常大的结果集。
如果函数无法连接到节点，则它们会提前出错。
可能会导致严重的结果。

SQL 支持和解决方法

由于 Citus 通过扩展 Spacture 提供分布式功能，因此它与 Spacture 兼容。这意味着用户可以将丰富且可扩展的 Spacture 生态系统分发的工具和功能用于使用 Citus 创建的分布式表。

对于能够在单个Worker节点上执行的任何查询，Citus 具有 100% 的 SQL 覆盖率。在访问有关单个租户的信息时，此类查询在多租户应用程序中很常见。

甚至跨节点查询（用于并行计算）也支持大多数 SQL 功能。但是，结合来自多个节点的查询不支持某些 SQL 功能。

跨节点 SQL 查询的限制：

只有在 PARTITION BY 中包含分布列时才支持窗口函数。
SELECT … FOR UPDATE 仅适用于单分片查询
TABLESAMPLE 仅适用于单分片查询
仅当关联在分布列上并且子查询符合子查询下推规则（例如，按分布列分组，没有 LIMIT 或 LIMIT OFFSET 子句）时，才支持关联子查询。
递归 CTE 仅适用于单分片查询
分组集仅适用于单分片查询

解决方法 (Workarounds)

Citus 支持多租户使用场景中的所有 SQL 语句。即使在实时分析使用场景中，查询跨越节点，Citus 也支持大多数语句。许多不受支持的功能都有解决方法；下面是一些最有用的。

JOIN本地表和分布式表(JOIN a local and a distributed table)

尝试在本地表“local”和分布式表“dist”之间执行 JOIN 会导致错误：

SELECT * FROM local JOIN dist USING (id);

/*
ERROR:  relation local is not distributed
STATEMENT:  SELECT * FROM local JOIN dist USING (id);
ERROR:  XX000: relation local is not distributed
LOCATION:  DistributedTableCacheEntry, metadata_cache.c:711
*/

虽然你不能直接连接这样的表，但通过将本地表包装在子查询或 CTE 中，你可以让 Citus 的递归查询planner将本地表数据复制到Worker节点。通过Co-locating数据，这允许查询继续进行。

-- either

SELECT *
  FROM (SELECT * FROM local) AS x
  JOIN dist USING (id);

-- or

WITH x AS (SELECT * FROM local)
SELECT * FROM x
JOIN dist USING (id);

Coordinator会将子查询或 CTE 中的结果发送给需要它进行处理的所有Worker节点。因此，最好尽可能向内部查询添加最具体的过滤条件和限制，或者聚合表。这减少了此类查询可能导致的网络开销。在子查询/CTE 网络开销中有更多关于此的信息。

临时表 (Temp Tables)：最后的解决方法

即使通过子查询使用push-pull执行，仍有一些查询不受支持。其中之一是运行按非分布列分区的窗口函数。

在我们的实时分析教程中，我们创建了一个名为 github_events 的表，由 user_id 列分布。如果我们想查询它并为预选的 repos 计算事件，在 SQL 中执行此操作的一种方法是使用窗口函数。但是，对非分布列进行分区不起作用：

-- this won't work
SELECT repo_id, org->'id' as org_id, count(*)
  OVER (PARTITION BY repo_id) __-- repo_id is not distribution column__
  FROM github_events
 WHERE repo_id IN (8514, 15435, 19438, 21692);

不过有一个技巧。我们可以将相关信息作为临时表拉到Coordinator节点中：

-- grab the data, minus the aggregate, into a local table

CREATE TEMP TABLE results AS (
  SELECT repo_id, org->'id' as org_id
    FROM github_events
   WHERE repo_id IN (8514, 15435, 19438, 21692)
);

-- now run the aggregate locally

SELECT repo_id, org_id, count(*)
  OVER (PARTITION BY repo_id)
  FROM results;

在Coordinator节点上创建临时表是最后的手段。它受节点的磁盘大小和 CPU 的限制。

Citus API

Citus 函数

本节包含 Citus 提供的用户定义函数的参考信息。除了标准 SQL 命令之外，这些函数有助于为 Citus 提供额外的分布式功能。

表和分片 DDL （Table and Shard DDL）

create_distributed_table

create_distributed_table() 函数用于定义分布式表, 并且如果它是哈希分布式表，还用于创建它的分片。此函数采用表名称、分布列和可选的分发方法，并插入适当的元数据以将表标记为分布式。如果未指定分配方法，则该函数默认为“哈希”分布。如果表是哈希分布式，则该函数还基于分片计数和分片复制因子配置值创建Worker节点分片。如果表包含任何行，会将这些行自动分布到Worker节点。

参数 (Arguments)

table_name：需要分布的表名称。

distribution_column：表分布到的列。

distribution_type：（可选）表分布依据的方法。允许值为“追加”或“哈希”，默认值为“哈希”。

colocate_with：（可选）将当前表包含在另一个表的Co-Location组中。默认情况下，当表是由同一类型的列分布、具有相同的分片计数，并且具有相同的复制因子时，将Co-Lcation这些表。 colocate_with 的可能值为 default、启动新Co-Location组的 none，或与该表Co-Lcation的另一个表的名称。

请记住，colocate_with 的默认值进行隐式Co-Location。在表之间相关或将互相连接时，Co-Location可能是很好的做法。但是，如果两个表不相关，但它们的分布列使用相同的数据类型，则意外Co-Location这些表会在分片rebalance期间降低性能。表分片将以"级联"方式不必要地一起移动。

如果新的分布式表与其他表不相关，最好指定 colocate_with ⇒ 'none'。

返回值

N/A

示例

此示例通知数据库：github_events 表应由哈希分布到 repo_id 列。

SELECT create_distributed_table('github_events', 'repo_id');

-- alternatively, to be more explicit:
SELECT create_distributed_table('github_events', 'repo_id',
                                colocate_with => 'github_repo');

create_reference_table

create_reference_table() 函数用于定义小引用表或维度表。此函数采用表名称，创建一个只包含一个分片的分布式表，并将其复制到每个Worker节点。

参数 (Arguments)

table_name：需要分布的小型维度表或引用表的名称。

返回值

N/A

示例

此示例通知数据库，国家表应定义为引用表

SELECT create_reference_table('nation');

upgrade_to_reference_table

upgrade_to_reference_table() 函数将使用一个现有分布式表（该表的分片计数为 1），并将其升级为可识别的引用表。调用此函数后，表就像是用 create_reference_table 创建的一样。

参数 (Arguments)

table_name：包含分片计数为 1 的分布式表的名称，将以引用表的形式分布。

返回值

N/A

示例

此示例通知数据库：国家表应定义为引用表

SELECT upgrade_to_reference_table('nation');

mark_tables_colocated

mark_tables_colocated() 函数采用分布式表（源表）和其他表的列表（目标表），并将目标表置于与源表相同的Co-Location组中。如果源表尚不在组中，则此函数将创建一个组，并为其分配源表和目标表。

应通过 create_distributed_table 的 colocate_with 参数在表分布时Co-Location表，但也可根据需要，稍后由 mark_tables_colocated Co-Location表。

参数 (Arguments)

source_table_name：目标将被分布到匹配的Co-Location组的分布式表的名称。

target_table_names：分布式目标表的名称数组，必须为非空。这些分布式表必须与下列因素中的源表匹配：

分布方式
分布列类型
复制类型
分片计数

如果上述条件均不适用，则Citus将产生错误。例如，尝试Co-Location表 apples 和 oranges，其分布列类型会产生不同的结果：

ERROR:  XX000: cannot colocate tables apples and oranges
DETAIL:  Distribution column types don't match for apples and oranges.

返回值

N/A

示例

此示例将 products 和 line_items 置于与 stores 相同的Co-Location组中。该示例假定这些表全部分布到具有匹配类型的列中，很可能是"存储 id"。

SELECT mark_tables_colocated('stores', ARRAY['products', 'line_items']);

master_create_empty_shard

master_create_empty_shard() 函数可用于为附加分布式表创建一个空分片。该函数首先选择 shard_replication_factor Worker节点来创建分片。然后，它连接到Worker节点并为所选Worker节点上的分片创建空的位置。最后，更新Coordinator节点上这些位置的元数据，使这些分片对未来的查询可见。如果无法创建所需数量的分片放置，则该函数会出错。

参数 (Arguments)

table_name：要为其创建新分片的附加分布式表的名称。

返回值

shard_id：该函数返回分配给新创建的分片的唯一 id。

示例

此示例为 github_events 表创建一个空分片。创建的shard的shard id为102089。

SELECT * from master_create_empty_shard('github_events');

master_create_empty_shard

-------------------------- -

                102089

(1 row)

表和分片 DML (Table and Shard DML)

master_append_table_to_shard

master_append_table_to_shard() 函数可用于将 Spacture 表的内容append到append分布式表的分片中。函数连接到每个具有该分片和该分片副本的Worker节点，并将表的内容附加到每个Worker节点。然后，该函数根据对每个分片位置的append是成功还是失败, 来更新分片位置的元数据。

如果该函数能够成功地append到至少一个分片放置，则该函数运行成功，并返回值。该函数同时还会将任何附加失败的分片标记为INACTIVE，这样之后的所有查询都不会考虑该分片。如果每个分片都append失败，那么该函数将退出并显示error（因为没有append任何数据）。在这种情况下，元数据保持不变。

参数 (Arguments)

shard_id：必须附加表内容的分片的 ID。

source_table_name：需要附加内容的 Spacture 表的名称。

source_node_name：存在源数据的节点（CN节点）的DNS名称

source_node_port：数据库服务器正在侦听的源Worker节点上的端口。

返回值

shard_fill_ratio：该函数返回分片的填充率，该填充率的定义为，当前分片大小与配置参数shard_max_size的比率。

示例

本示例将github_events_local表的内容追加到ID为102089的分片上。表github_events_local位于运行在节点master-101上的端口号5432上的数据库中。该函数返回当前分片大小与对分片的比率。最大分片大小，为0.1，表示已填充10％的分片。

SELECT * from
master_append_table_to_shard(102089,'github_events_local','master-101', 5432);

master_append_table_to_shard

------------------------------

                 0.100548

(1 row)

master_apply_delete_command

master_apply_delete_command() 函数用于删除与附加分布式表上的删除命令指定的条件匹配的分片。仅当分片中的所有行都符合删除条件时，此函数才会删除分片。由于该函数使用分片元数据来决定是否需要删除分片，因此它要求 DELETE 语句中的 WHERE 子句位于分布列上。如果未指定任何条件，则删除该表的所有分片。

此功能连接到所有具有与删除标准匹配的分片的Worker节点，并向它们发送命令以删除所选分片。然后，该函数更新Coordinator节点上的相应元数据。如果该函数能够成功删除分片放置，则其元数据将被删除。如果无法删除特定展示位置，则将其标记为要删除。标记为 TO DELETE 的展示位置不会被考虑用于以后的查询，并且可以稍后进行清理。

参数 (Arguments)

delete_command：有效的 SQL DELETE 命令

返回值

Deleted_shard_count：该函数返回符合条件并被删除（或标记为删除）的分片数量。请注意，这是分片(shard)的数量（pg_dist_shard），而不是分片放置(shard placement)的数量(pg_dist_shard_plactment)。

示例

第一个示例删除 github_events 表的所有分片，因为没有指定删除条件。在第二个示例中，仅删除符合条件的分片（在本例中为 3）。

SELECT * from master_apply_delete_command('DELETE FROM github_events');

master_apply_delete_command

-----------------------------

                           5
(1 row)

SELECT * from master_apply_delete_command('DELETE FROM github_events
WHERE review_date < ''2009-03-01''');

[source,sql]
 master_apply_delete_command

-----------------------------

                           3
(1 row)

master_modify_multiple_shards

master_modify_multiple_shards() 函数用于运行可以跨越多个分片的数据修改语句。根据 citus.multi_shard_commit_protocol 的值，提交可以分一阶段或两阶段完成。

局限性：

不能在事务块内调用
只能使用简单的运算符表达式调用它

参数 (Arguments)

modify_query：一个简单的DELETE或UPDATE查询字符串。

返回值

N/A

示例

SELECT master_modify_multiple_shards(

  'DELETE FROM customer_delete_protocol WHERE c_custkey > 500 AND c_custkey < 500');

元数据/配置信息

master_add_node

master_add_node() 函数在 Citus 元数据表 pg_dist_node 中的集群中注册新节点添加。它还将引用表复制到新节点。

参数 (Arguments)

node_name：要添加的新节点的 DNS 名称或 IP 地址。

node_port：Spacture 在Worker节点上侦听的端口。

group_id：一组一个主服务器和零个或多个辅助服务器，仅与流复制相关。默认 0

node_role：是‘primary’还是‘secondary’。默认 ‘primary’。

node_cluster：集群名称。默认‘default’

返回值

代表 pg_dist_node 表中一行的元组

示例

select * from master_add_node('new-node', 12345);

 nodeid | groupid | nodename | nodeport | noderack | hasmetadata | isactive | groupid | noderole | nodecluster

--------+---------+----------+----------+----------+-------------+----------+---------+-----------+------------

      7 |       7 | new-node |     12345 | default  | f           | t       |       0 | primary  | default

(1 row)

master_update_node

master_update_node() 函数更改在 Citus 元数据表 pg_dist_node 中注册的节点的主机名和端口。

参数 (Arguments)

node_id：来自 pg_dist_node 表的 id。

node_name：更新节点的 DNS 名称或 IP 地址。

node_port：Spacture 在Worker节点上侦听的端口。

返回值

N/A

示例

select * from master_update_node(123, 'new-address', 5432);

master_add_inactive_node

master_add_inactive_node 函数，类似于 master_add_node，在 pg_dist_node 中注册一个新节点。但是它将新节点标记为不活动，这意味着不会在那里放置分片。此外，它不会将引用表复制到新节点。

参数 (Arguments)

node_name：要添加的新节点的 DNS 名称或 IP 地址。

node_port：Spacture 在Worker节点上侦听的端口。

group_id：一组一个主服务器和零个或多个辅助服务器，仅与流复制相关。默认 0

node_role：是‘primary’还是‘secondary’。默认 ‘primary’。

node_cluster：集群名称。默认‘default’

返回值

代表 pg_dist_node 表中一行的元组

示例

select * from master_add_inactive_node('new-node', 12345);

 nodeid | groupid | nodename | nodeport | noderack | hasmetadata |isactive | groupid | noderole | nodecluster

--------+----------+----------+----------+----------+-------------+----------+---------+----------+-------------

      7 |        7 | new-node |    12345 | default  | f           | f        |       0 | primary  | default

(1 row)

master_activate_node

master_activate_node 函数在 Citus 元数据表 pg_dist_node 中将节点标记为活动节点，并将引用表复制到该节点。对通过 master_add_inactive_node 添加的节点很有用。

参数 (Arguments)

node_name：要添加的新节点的 DNS 名称或 IP 地址。

node_port：Spacture 在Worker节点上侦听的端口。

返回值

代表 pg_dist_node 表中一行的元组

示例

select * from master_activate_node('new-node', 12345);

 nodeid | groupid | nodename | nodeport | noderack | hasmetadata |isactive| noderole | nodecluster

--------+---------+----------+----------+----------+-------------+---------+----------+-------------

      7 |       7 | new-node |    12345 | default  | f           | t       | primary  | default

(1 row)

master_disable_node

master_disable_node 功能与 master_activate_node 功能相反。它在 Citus 元数据表 pg_dist_node 中将节点标记为不活动，暂时将其从集群中删除。该函数还会从禁用的节点中删除所有参考表放置。要重新激活节点，只需再次运行 master_activate_node。

参数 (Arguments)

node_name：要禁用的节点的 DNS 名称或 IP 地址。

node_port：Spacture 在Worker节点上侦听的端口。

返回值

N/A

示例

select * from master_disable_node('new-node', 12345);

master_add_secondary_node

master_add_secondary_node() 函数在节点中为现有的主节点注册一个新的辅助节点。它更新 Citus 元数据表 pg_dist_node。

参数 (Arguments)

node_name：要添加的新节点的 DNS 名称或 IP 地址。

node_port：Spacture 在Worker节点上侦听的端口。

primary_name：此辅助节点的主节点的 DNS 名称或 IP 地址。

primary_port：Spacture 在主节点上监听的端口。

node_cluster：节点名称。默认‘default’

返回值

代表 pg_dist_node 表中一行的元组。

示例

select * from master_add_secondary_node('new-node', 12345,'primary-node', 12345);

 nodeid | groupid | nodename | nodeport | noderack | hasmetadata |isactive | noderole  | nodecluster

--------+---------+----------+----------+----------+-------------+---------+-----------+-------------

      7 |       7 | new-node |    12345 | default  | f           | t       | secondary | default

(1 row)

master_remove_node

master_remove_node() 函数从 pg_dist_node 元数据表中删除指定的节点。如果此节点上存在现有的分片放置，则此函数将出错。因此，在使用此功能之前，需要将分片移出该节点。

参数 (Arguments)

node_name：要删除的节点的 DNS 名称。

node_port：Spacture 在Worker节点上侦听的端口。

返回值

N/A

示例

select master_remove_node('new-node', 12345);

master_remove_node

------------------- -

(1 row)

master_get_active_worker_nodes

master_get_active_worker_nodes() 函数返回活动Worker节点主机名和端口号的列表。

参数 (Arguments)

N/A

返回值

元组列表，其中每个元组包含以下信息：

node_name：Worker节点的 DNS 名称

node_port：数据库服务器正在侦听的Worker节点上的端口

示例

SELECT * from master_get_active_worker_nodes();

 node_name | node_port

-----------+-----------

 localhost |      9700

 localhost |      9702

 localhost |      9701

(3 rows)

master_get_table_metadata

master_get_table_metadata() 函数可用于为分布式表返回与分布相关的元数据。此元数据包括表的关系 ID、存储类型、分布方法、分布列、复制计数、最大分片大小和分片放置策略。除了这些内容，此函数将查询Citus元数据表，以获取所需的信息，并将其连接到元组，然后将其返回给用户。

参数 (Arguments)

table_name：要为其提取元数据的分布式表的名称。

返回值

包含以下信息的元组：

logical_relid：分布式表的 Oid。它引用 pg_class 系统目录表中的 relfilenode 列。

part_storage_type：用于表的存储类型。可以是“t”（标准表）、“f”（外表）或“c”（柱状表）。

part_method：用于表的分布方法。可以是“a”（追加）或“h”（哈希）。

part_key：表的分布列。

part_replica_count：当前分片复制计数。

part_max_size：当前最大分片大小（以字节为单位）。

part_placement_policy：用于放置表的分片的分片放置策略。可以是 1（本地节点优先）或 2（轮循机制）。

示例

下面的示例提取并显示 github_events 表的表元数据。

SELECT * from master_get_table_metadata('github_events');

 logical_relid | part_storage_type | part_method | part_key |part_replica_count | part_max_size | part_placement_policy

---------------+-------------------+-------------+----------+--------------------+---------------+-----------------------

         24180 | t                 | h           | repo_id  |                 2        |   1073741824 |                     2

(1 row)

get_shard_id_for_distribution_column

Citus根据行的分布列的值和表的分布方法，将分布式表的每一行分布给分片。在大多数情况下，精确映射是数据库管理员可忽略的低级别详细信息。然而，无论是对于手动数据库维护任务，还是为了确认数据分布，确定行的分片会很有用。 get_shard_id_for_distribution_column 函数提供用于哈希分布式表、范围分布式表以及引用表的信息。它不适用于追加分布。

参数 (Arguments)

table_name：分布式表。

distribution_value：分布列的值

返回值

分片 ID Citus 与给定表的分布列值相关联。

示例

SELECT get_shard_id_for_distribution_column('my_table', 4);

get_shard_id_for_distribution_column

------------------------------------- -

                               540007

(1 row)

column_to_column_name

将 pg_dist_partition 的 partkey 列转换为文本列名称。该转换可用于确定分布式表的分布列。

参数 (Arguments)

table name：分布式表。_

column_var_text：pg_dist_partition 表中的 partkey 值。

返回值

table_name 的分布列名称。

示例

-- get distribution column name for products table

SELECT column_to_column_name(logicalrelid, partkey) AS dist_col_name
  FROM pg_dist_partition
 WHERE logicalrelid='products'::regclass;

输出

┌───────────────┐
│ dist_col_name │
├───────────────┤
│ company_id    │
└───────────────┘

citus_relation_size

获取指定的分布式表的所有分片所使用的磁盘空间。磁盘空间包括" main fork"的大小，但不包括分片的可见性映射(vm)和可用空间映射(fsm)。

参数 (Arguments)

logicalrelid：分布式表的名称。

返回值

以字节为单位的大小，为 bigint。

示例

SELECT pg_size_pretty(citus_relation_size('github_events'));


pg_size_pretty

------------- -

23 MB

citus_table_size

获取指定的分布式表的所有分片所使用的磁盘空间，不包括索引，包括 TOAST、可用空间映射(fsm)和可见性映射(vm）。

参数 (Arguments)

logicalrelid：分布式表的名称。

返回值

以字节为单位的大小，为 bigint。

示例

SELECT pg_size_pretty(citus_table_size('github_events'));


pg_size_pretty

--------------

37 MB

citus_total_relation_size

获取指定的分布式表的所有分片使用的总磁盘空间，包括所有索引和 TOAST 数据。

参数 (Arguments)

logicalrelid：分布式表的名称。

返回值

以字节为单位的大小，为 bigint。

示例

SELECT pg_size_pretty(citus_total_relation_size('github_events'));


pg_size_pretty

--------------

73 MB

citus_stat_statements_reset

从 citus_stat_statements 中移除所有行。此函数独立于 pg_stat_statements_reset() 使用。如果要重置所有统计信息，请调用这两个函数。

参数 (Arguments)

N/A

返回值

无

服务器组管理和修复

master_copy_shard_placement

如果在修改命令或 DDL 操作过程中无法更新分片放置，则会将其标记为非活动状态。然后，可以调用 master_copy_shard_placement 函数来通过使用正常放置的数据修复非活动的分片放置。

如果要修复分片，该函数将首先删除不正常的分片放置，并使用Coordinator节点上的架构重新创建。创建分片放置后，该函数将从正常放置复制数据，并更新元数据以将新的分片放置标记为正常。此函数可确保在修复过程中，避免对分片进行任何并发修改。

参数 (Arguments)

shard_id：要修复的分片的 ID。

source_node_name：存在正常分片放置的节点的 DNS 名称（"源"节点）。

source_node_port：数据库服务器正在侦听的源Worker节点上的端口。

target_node_name：存在无效分片放置的节点的 DNS 名称（"目标"节点）。

target_node_port：数据库服务器正在侦听的目标Worker节点上的端口。

返回值

N/A

示例

下面的示例将修复分片 12345 的非活动分片放置，该分片放置位于在端口 5432 上“bad_host”中运行的数据库服务器上。如果要修复此问题，它将使用以下端口上“good_host”中运行的服务器上的正常分片放置的数据： 5432.

SELECT master_copy_shard_placement(12345, 'good_host', 5432, 'bad_host',
5432);

Citus 表和视图 (Citus Tables and Views)

Coordinator节点元数据

Citus会基于分布列将每个分布式表划分为多个逻辑分片。 Coordinator节点随后会维护元数据表，以跟踪有关这些分片的运行状况和位置的统计信息和信息。在本节中，我们将描述每个元数据表及其架构。通过Spacture客户端连接Coordinator节点后，可以使用 SQL 查看和查询这些表。

分区表

pg_dist_partition 表存储有关数据库中的分布式表的元数据。对于每个分布式表，它还存储有关分布方法的信息以及有关分布列的详细信息。

名称	类型	说明
logicalrelid	regclass	此行对应的分布式表。此值引用pg-class系统目录表中的relfilenode 列
partmethod	char	用于分区/分布的方法，与不同分布方法对应的此列值是追加：‘a’, 哈希: ‘h’, 引用表: ‘n’
partkey	text	有关分布列的详细信息，包括列号、类型和其他信息。
colocationid	integer	此表所属的归置组。同一组中的表允许在其他优化之间实现归置联接和分布式汇总。此值引用pg_dist_colocation 表中的 colocationid 列。
repmodel	char	用于数据复制的方法。此列与不同复制方法对应的值包括：Citus基于语句的复制: “c”，Spacture流式处理复制: “s”, 两阶段提交（适用于引用表）：“t”

名称

类型

说明

logicalrelid

regclass

此行对应的分布式表。此值引用pg-class系统目录表中的relfilenode 列

partmethod

char

用于分区/分布的方法，与不同分布方法对应的此列值是追加：‘a’, 哈希: ‘h’, 引用表: ‘n’

partkey

text

有关分布列的详细信息，包括列号、类型和其他信息。

colocationid

integer

此表所属的归置组。同一组中的表允许在其他优化之间实现归置联接和分布式汇总。此值引用pg_dist_colocation 表中的 colocationid 列。

repmodel

char

用于数据复制的方法。此列与不同复制方法对应的值包括：Citus基于语句的复制: “c”，Spacture流式处理复制: “s”, 两阶段提交（适用于引用表）：“t”

SELECT * from pg_dist_partition;

logicalrelid  | partmethod |                                                       partkey                                                        | colocationid |
repmodel

---------------+------------+--------------------------------------------------------------------------------------------------------------------------------+---------------+----------

 github_events | h          | \{VAR :varno 1 :varattno 4 :vartype 20:vartypmod -1:varcollid 0 :varlevelsup 0 :varnoold 1 :varoattno 4:location -1} |            2 | c

 (1 row)

分片表

pg_dist_shard 表存储有关表的各个分片的元数据。 Pg_dist_shard 包含有关分片所属的分布式表的信息，以及有关分片的分布列的统计信息。对于追加分布式表，这些统计信息对应于分布列的最小值/最大值。对于哈希分布式表，它们是分配给该分片的哈希标记范围。这些统计信息用于在 SELECT 查询期间删除不相关的分片。

名称	类型	说明
logicalrelid	regclass	此行对应的分布式表。此值引用pg-class系统目录表中的relfilenode 列
shardid	bigint	分配给此分片的全局唯一标识符。
shardstorage	char	用于此分片的存储类型。下表讨论了不同的存储类型。
shardminvalue	text	对于追加分布式表，是此分片中分布列的最小值（含）。对于哈希分布式表，是分配给该分片的最小哈希标记值（含）。
shardmaxvalue	text	对于追加分布式表，是此分片中分布列的最大值（含）。对于哈希分布式表，是分配给该分片的最大哈希标记值（含）。

名称

类型

说明

logicalrelid

regclass

此行对应的分布式表。此值引用pg-class系统目录表中的relfilenode 列

shardid

bigint

分配给此分片的全局唯一标识符。

shardstorage

char

用于此分片的存储类型。下表讨论了不同的存储类型。

shardminvalue

text

对于追加分布式表，是此分片中分布列的最小值（含）。对于哈希分布式表，是分配给该分片的最小哈希标记值（含）。

shardmaxvalue

text

对于追加分布式表，是此分片中分布列的最大值（含）。对于哈希分布式表，是分配给该分片的最大哈希标记值（含）。

SELECT * from pg_dist_shard;

 logicalrelid  | shardid | shardstorage | shardminvalue | shardmaxvalue

---------------+---------+--------------+---------------+---------------

 github_events |  102026 | t            | 268435456     | 402653183

 github_events |  102027 | t            | 402653184     | 536870911

 github_events |  102028 | t            | 536870912     | 671088639

 github_events |  102029 | t            | 671088640     | 805306367

 (4 rows)

分片存储类型

pg_dist_shard 中的 shardstorage 列指示用于分片的存储类型。下面是不同分片存储类型及其表示形式的简要概述。

存储类型

Shardstorage值

说明

TABLE

“t”

指示分片存储属于常规分布式表的数据。

COLUMNAR

“c”

指示分片存储纵栏式数据。（由分布式cstore_fdw表使用）

FOREIGN

“f”

指示分片存储外部数据。（由分布式file_fdw表使用）

分片放置表

pg_dist_placement 表跟踪分片副本在Worker节点上的位置。分配给特定节点的分片的每个副本称为分片放置。此表存储有关每个分片放置的运行状况和位置的信息。

名称	类型	说明
shardid	bigint	与此防止关联的分片标识符。此值引用pg_dist_shard目录表中的shardid列。
shardstate	int	描述此放置的状态。以下部分讨论了不同的分片状态。
shardlength	bigint	对于追加分布式表，是worker节点上的分片放置大小（以字节为单位）。对于哈希分布式表，为零。
placementid	bigint	为每个单独放置自动生成的唯一标识符。
groupid	int	使用流式处理复制模型时，表示由一台主服务器和零台或多台辅助服务器组成的组。

名称

类型

说明

shardid

bigint

与此防止关联的分片标识符。此值引用pg_dist_shard目录表中的shardid列。

shardstate

int

描述此放置的状态。以下部分讨论了不同的分片状态。

shardlength

bigint

对于追加分布式表，是worker节点上的分片放置大小（以字节为单位）。对于哈希分布式表，为零。

placementid

bigint

为每个单独放置自动生成的唯一标识符。

groupid

int

使用流式处理复制模型时，表示由一台主服务器和零台或多台辅助服务器组成的组。

SELECT * from pg_dist_placement;

  shardid | shardstate | shardlength | placementid | groupid

 ---------+------------+-------------+-------------+---------

   102008 |          1 |           0 |           1 |       1

   102008 |          1 |           0 |           2 |       2

   102009 |          1 |           0 |           3 |       2

   102009 |          1 |           0 |           4 |       3

   102010 |          1 |           0 |           5 |       3

   102010 |          1 |           0 |           6 |       4

   102011 |          1 |           0 |           7 |       4

分片放置状态

Citus 按每个放置管理分片运行状况。如果放置使系统处于不一致状态，则 Citus 会自动将它标记为不可用。放置状态记录在 pg_dist_shard_placement 表的 shardstate 列中。下面是不同分片放置状态的简要概述：

状态名称	Shardstate值	说明
FINALIZED	1	新分片创建时所处的状态。此状态下的分片放置被视为是最新的，再查询规划和执行中使用。
INACTIVE	3	由于与同一分片的其他副本不同步，因此该状态下的分片放置被视为不活动。此放置的追加、修改 (INSERT, UPDATE or DELETE ) 或 DDL 操作失败时，可能会出现这种状态。查询规划器将在规划和执行期间忽略此状态的放置。用户可以采用后台活动的形式，将这些分片中的数据与最终副本同步。
TO_DELETE	4	如果 Citus 尝试删除分片放置以响应 master_apply_delete_command 调用，但是失败，则放置将转变为此状态。然后，用户可以采用后续后台活动的形式删除这些分片。

状态名称

Shardstate值

说明

FINALIZED

新分片创建时所处的状态。此状态下的分片放置被视为是最新的，再查询规划和执行中使用。

INACTIVE

由于与同一分片的其他副本不同步，因此该状态下的分片放置被视为不活动。此放置的追加、修改 (INSERT, UPDATE or DELETE ) 或 DDL 操作失败时，可能会出现这种状态。查询规划器将在规划和执行期间忽略此状态的放置。用户可以采用后台活动的形式，将这些分片中的数据与最终副本同步。

TO_DELETE

如果 Citus 尝试删除分片放置以响应 master_apply_delete_command 调用，但是失败，则放置将转变为此状态。然后，用户可以采用后续后台活动的形式删除这些分片。

Worker节点表

pg_dist_node 表包含有关集群中Worker节点的信息

名称	类型	说明
nodeid	int	为单个节点自动生成的标识符。
groupid	int	使用流式处理复制模型时，用于表示由一台主服务器和零台辅助服务器组成的组的标识符。默认情况下，它与nodeid相同。
nodename	text	Spacture worker节点的主机名或IP地址。
nodeport	int	Spacture worker节点正在侦听的端口号。
noderack	text	（可选）Worker节点的机架放置信息。
hasmetadata	boolean	保留以供内部使用。
isactive	boolean	节点是否处于活动状态，在接受分片放置。
noderole	text	节点是主节点还是辅助节点。
nodecluster	text	包含此节点的集群名称。

名称

类型

说明

nodeid

int

为单个节点自动生成的标识符。

groupid

int

使用流式处理复制模型时，用于表示由一台主服务器和零台辅助服务器组成的组的标识符。默认情况下，它与nodeid相同。

nodename

text

Spacture worker节点的主机名或IP地址。

nodeport

int

Spacture worker节点正在侦听的端口号。

noderack

text

（可选）Worker节点的机架放置信息。

hasmetadata

boolean

保留以供内部使用。

isactive

boolean

节点是否处于活动状态，在接受分片放置。

noderole

text

节点是主节点还是辅助节点。

nodecluster

text

包含此节点的集群名称。

SELECT * from pg_dist_node;

 nodeid | groupid | nodename  | nodeport | noderack | hasmetadata | isactive | noderole | nodecluster

--------+---------+-----------+----------+----------+-------------+----------+----------+-------------

      1 |       1 | localhost |    12345 | default  | f           | t        | primary  | default

      2 |       2 | localhost |    12346 | default  | f           | t        | primary  | default

      3 |       3 | localhost |    12347 | default  | f           | t        | primary  | default

(3 rows)

Co-Location组表

pg_dist_colocation 表包含有关表中应放置在一起或进行Co-Location的分片的信息。当两个表处于同一个Co-Location组中时，Citus 会确保将具有相同分区值的分片放置在同一个Worker节点上。 Co-Location可实现连接优化、特定分布式Rollup和外键支持。当两个表之间的分片计数、复制因子和分区列类型全部匹配时，会推断进行分片Co-Location；但是，如果需要，可以在创建分布式表时指定自定义Co-Location组。

名称

类型

说明

colocationid

int

此行对应的co-location 组的唯一标识符。

shardcount

int

此co-location 组中的所有表的分片计数。

replicationfactor

int

此co-location 组中的所有表的复制因子。

distributioncolumntype

oid

此co-location 组中所有表的分布式列的类型。

SELECT * from pg_dist_colocation;

  colocationid | shardcount | replicationfactor | distributioncolumntype

 --------------+------------+-------------------+------------------------

             2 |         32 |                 2 |                     20

  (1 row)

配置参考 (Configuration Reference)

有各种服务器参数会影响Citus 的行为，这些参数既包括标准 Spacture 中的参数，也包括特定于 Citus 的参数。

本参考的其余部分旨在讨论 Citus 特定的配置参数。可以通过修改 postgresql.conf 或使用 SET 命令来设置这些参数，类似于Spacture参数。

例如，可以使用以下命令更新设置：

ALTER DATABASE citus SET citus.multi_task_query_log_level = 'log';

常规配置

citus.max_worker_nodes_tracked（整数）

Citus 在Coordinator节点上的共享哈希表中跟踪Worker节点的位置及其成员资格。此配置值限制了哈希表的大小，从而限制了可以跟踪的Worker节点的数量。此设置的默认值为 2048。此参数只能在服务器启动时设置，并且在Coordinator节点上有效。

citus.use_secondary_nodes（枚举）

设置选择 SELECT 查询的节点时要使用的策略。如果设置为“always”，则planner仅查询 pg_dist_node 中标记为“secondary”节点角色的节点。

此枚举支持的值为：

never：（默认值）所有读取都在主要节点上发生。
always：读取针对secondary节点运行，而插入/更新语句将会禁用。

citus.cluster_name（文本）

告知Coordinator节点planner它要协调哪个群集。设置 cluster_name 后，planner只会查询该群集中的Worker节点。

citus.enable_version_checks（布尔值）

升级Citus 版本需要重启服务器（以使用新版本的动态链接库），然后运行 ALTER EXTENSION UPDATE 命令。执行这两个步骤失败可能会导致出错或崩溃。 Citus 因而会验证代码版本和扩展版本是否匹配，如果不匹配，则会出错。

该值默认为 true，并在Coordinator节点上生效。在极少数情况下，复杂的升级过程可能要求将此参数设置为 false，从而禁用检查。

citus.log_distributed_deadlock_detection（布尔值）

是否在服务器日志中记录分布式死锁检测相关的处理。默认值为 false。

citus.distributed_deadlock_detection_factor（浮点值）

设置在检查分布式死锁之前要等待的时间。具体而言，等待时间是将此值与 Spacture 的 deadlock_timeout 设置相乘后得到的结果。默认值为 2。指定 -1 值会禁用分布式死锁检测。

citus.node_conninfo（文本）

citus.node_conninfo GUC 设置用于所有节点间连接的非敏感 libpq 连接参数。

key=value pairs separated by spaces.

For example, ssl options:

ALTER DATABASE foo

SET citus.node_conninfo =

  'sslrootcert=/path/to/citus.crt sslmode=verify-full';

Citus 仅支持列入白名单的选项子集，即：

application_name
connect_timeout
gsslib†
keepalives
keepalives_count
keepalives_idle
keepalives_interval
krbsrvname†
sslcompression
sslcrl
sslmode (自 Citus 8.1 起默认为“require”)
sslrootcert

(† = 取决于可选 Spacture 功能的运行时存在)

数据加载

citus.multi_shard_commit_protocol（枚举）

设置对哈希分布式表执行COPY时要使用的提交协议。在每个单独的分片placement上，COPY将在事务块中执行，以确保在COPY期间发生error时不会摄取任何数据。但存在一种特殊的error情况, 即 COPY 在所有placement上都成功，但在提交所有事务之前发生（硬件）error。使用此参数可在以下提交协议之间进行选择，以防止在出现这种情况时丢失数据：

2pc：（default）首先使用 Spacture 的两阶段提交来准备在分片位置执行复制时所在的事务，然后提交这些事务。可以使用 COMMIT PREPARED 或 ROLLBACK PREPARED 来分别手动恢复或中止失败的提交。使用 2pc 时，应在所有Worker上增大 max_prepared_transactions，通常应将设置为与 max_connections 相同的值。
1pc：通过一个轮次提交在分片位置执行复制时所在的事务。如果在所有位置复制成功后提交失败，则可能会丢失数据（这种情况很罕见）。

citus.shard_replication_factor（整数）

设置分片的复制因子，即，要将分片放置到的节点数，默认值为 1。此参数可在运行时设置，并在Coordinator上生效。此参数的理想值取决于群集大小和节点故障率。例如，如果你运行大型群集并发现节点故障更频繁发生，则可能需要增大此复制因子。

citus.shard_count（整数）

为哈希分区表设置分片计数，默认值为 32。创建哈希分区表时，此值将由 create_distributed_table UDF使用。此参数可在运行时设置，并在Coordinator上生效。

citus.shard_max_size（整数）

设置分片在被拆分之前要增长到的最大大小，默认值为 1 GB。如果一个分片的源文件大小（用于暂存）超过此配置值，数据库会确保创建新分片。此参数可在运行时设置，并在Coordinator上生效。

planner配置

citus.limit_clause_row_fetch_count（整数）

设置每个任务要提取的行数以进行 limit 子句优化。在某些情况下，包含 limit 子句的 select 查询可能需要从每个任务中提取所有行来生成结果。在这种情况下，以及当近似值会生成有意义的结果时，此配置值将设置要从每个分片中提取的行数。 default会禁用 limit 近似值，此参数设置为 -1。此值可在运行时设置，并在Coordinator上生效。

citus.count_distinct_error_rate（浮点值）

Citus可以使用 postgresql-hll 扩展计算 count(distinct) 近似值。在计算 count(distinct) 时，此配置条目将设置所需的错误率。指定 0.0（默认值）会禁用 count(distinct) 的近似值；指定 1.0 不能保证结果的正确性。建议将此参数设置为 0.005 以获得最佳结果。此值可在运行时设置，并在Coordinator上生效。

citus.task_assignment_policy（枚举）

Note: 仅当 shard_replication_factor 大于 1 时，此 GUC 才适用；或者，此 GUC 仅适用于针对 reference_tables 运行的查询。

设置将任务分配给Worker时要使用的策略。 Coordinator节点基于分片位置将任务分配给Worker节点。此配置值指定进行这些分配时要使用的策略。目前可以使用三个可能的任务分配策略。

greedy：greedy策略是默认策略，旨在将任务均匀分配到Worker节点。
round-robin：round-robin以轮循方式将任务分配到Worker节点（在不同的副本之间交替分配）。当表的分片计数小于Worker节点数量时，此策略可以提高群集利用率。
first-replica：first-replica策略基于分片的位置（副本）插入顺序分配任务。换言之，对某个分片的片段查询将分配到包含该分片的第一个副本的Worker节点。使用此方法可以在有关要在哪些节点上使用哪些分片方面获得可靠的保证（即，获取更可靠的内存驻留保证）。

此参数可在运行时设置，并在Coordinator节点上生效。

中间结果数据传输

citus.binary_worker_copy_format（布尔值）

使用二进制复制格式在Worker节点之间传输中间数据。在连接大型表的过程中，Citus 可能需要动态重新分区，并在不同的Worker节点之间随机排布数据。默认情况下，将以文本格式传输此数据。启用此参数会指示数据库使用 Spacture 的二进制序列化格式传输此数据。此参数在Worker节点上生效，需要在 spacture 文件中进行更改。编辑配置文件后，用户可以发送 SIGHUP 信号或重启服务器，使此更改生效。

citus.binary_master_copy_format（布尔值）

使用二进制复制格式在Coordinator节点与Worker节点之间传输数据。运行分布式查询时，Worker节点会将其中间结果传输到Coordinator节点进行最终聚合。默认情况下，将以文本格式传输此数据。启用此参数会指示数据库使用 Spacture 的二进制序列化格式传输此数据。此参数可在运行时设置，并在Coordinator节点上生效。

citus.max_intermediate_result_size（整数）

无法推送到Worker节点以供执行的 CTE 以及复杂子查询的中间结果的最大大小 (KB)。默认值为 1 GB，值 -1 表示无限制。超过限制的查询将被取消并报错。

DDL

citus.enable_ddl_propagation（布尔值）

指定是否将Coordinator节点中发生的 DDL 更改自动分发到所有Worker节点。默认值为 true。由于某些架构更改需要在表中设置一个访问排他锁，并且自动分发序应用于所有Worker节点，因此它可能会导致Citus 群集的响应能力暂时降低。你可以选择禁用此设置并手动传播更改。

Executor配置

常规

citus.all_modifications_commutative

Citus强制实施交换性规则，并获取修改操作的相应锁以保证行为的正确性。例如，假设一个 INSERT 语句可与另一个 INSERT 语句交换，但不可与某个 UPDATE 或 DELETE 语句交换。同理，假设一个 UPDATE 或 DELETE 语句不可与另一个 UPDATE 或 DELETE 语句交换。这种防备意味着 UPDATE 和 DELETE 语句要求 Citus 获取更强的锁。

如果你的 UPDATE 语句可与 INSERT 或其他 UPDATE 语句交换，则你可以通过将此参数设置为 true 来放宽这些交换性假设。当此参数设置为 true 时，所有命令将被视为可交换并声明一个共享锁，这可以提高总体吞吐量。此参数可在运行时设置，并在Coordinator上生效。

citus.max_task_string_size（整数）

设置Worker节点任务调用字符串的最大大小（以字节为单位）。更改此值需要重新启动服务器，无法在运行时更改。

在主节点上的共享哈希表中跟踪活动的工作任务。此配置值限制单个工作任务的最大大小，并影响预分配共享内存的大小。

最小值：8192，最大值 65536，默认 12288

citus.remote_task_check_interval（整数）

设置 Citus 检查task-tracker所管理作业的状态的频率。默认值为 10 毫秒。 Coordinator节点将任务分配给Worker节点，然后定期在Worker节点上检查每个任务的进度。此配置值设置每两次后续检查的时间间隔。此参数在Coordinator上生效，可在运行时设置。

citus.task_executor_type（枚举）

Citus 提供两种executor类型用于运行分布式 SELECT 查询。可以通过设置该配置参数来选择所需的executor。此参数的可接受值为：

real-time：real-time是默认executor，当需要快速响应涉及跨多个分片的聚合和Co-Location连接的查询时，它是最佳选择。
task-tracker：task-tracker非常适合用于需要跨Worker节点随机排布数据的长时间运行的复杂查询，以及实现高效的资源管理。

此参数可在运行时设置，并在Coordinator节点上生效。

citus.multi_task_query_log_level（枚举）

为生成多个任务（即，命中多个分片）的任何查询设置日志级别。在迁移多租户应用程序期间，日志记录非常有用，因为你可以选择对此类查询显示错误或警告、查找这些查询并对其添加 tenant_id 筛选器。此参数可在运行时设置，并在Coordinator 节点上生效。此参数的默认值为 'off'。

此枚举支持的值为：

off：禁用记录任何生成多个任务（即，命中多个分片）的查询
debug：在“DEBUG”严重性级别记录语句。
log：在“LOG”严重性级别记录语句。日志行将包含运行的 SQL 查询。
notice：在“NOTICE”严重性级别记录语句。
warning：在“WARNING”严重性级别记录语句。
error：在“ERROR”严重性级别记录语句。

请注意，有利的用法可能是在开发测试期间使用 error，在实际生产部署期间使用更低的日志级别，例如 log。选择 log 会导致多任务查询显示在数据库日志中，查询本身显示在 "STATEMENT" 之后。

LOG: multi-task query about to be executed

HINT: Queries are split to multiple tasks if they have to be split into several queries on the workers.

STATEMENT: select * from foo;

citus.propagate_set_commands（枚举）

确定哪些 SET 命令从Coordinator节点分发到Worker节点。此参数的默认值为“none”。

支持的值为：

none：不传播 SET 命令。
local：仅传播 SET LOCAL 命令。

citus.enable_repartition_joins（布尔值）

一般情况下，尝试使用 adaptive executor执行repartition join会失败并出现错误消息。但是，将 citus.enable_repartition_joins 设置为 true 可使Citus 能够暂时切换到 task-tracker来执行连接。默认值为 false。

Adaptive executor配置

citus.max_adaptive_executor_pool_size（整数）

adaptive executor用于执行多分片命令的每个Worker节点的最大连接数。

adaptive executor在运行多分片命令时为每个Worker节点打开多个连接，以在Worker节点上的多个cpu上并行执行查询。此设置指定它将打开的最大连接数。（连接数也受节点上分片数的限制。）

虽然增加单个查询的并行度通常是好的，但它也会给生产环境中的Worker节点带来连接压力。该 GUC来权衡并行度与连接压力。

默认值是 16，但是很难手动做出好的静态选择。未来，adaptive executor可能会动态调整这个 GUC，在Worker节点忙碌时缩小池，在空闲时扩大池。

这是一个会话级别的 GUC，因此可用于限制单个会话的资源使用。降低该值可以限制查询对连接、CPU或内存的使用。

citus.executor_slow_start_interval（整数）

与同一Worker节点建立连接之间的等待时间（以毫秒为单位）。

当多分片查询的单个任务花费很少的时间时，它们通常可以通过单个（通常已经缓存）连接完成。为了避免冗余地打开额外的连接，executor在连接尝试之间等待配置的毫秒数。在间隔结束时，它会增加下一次允许打开的连接数。

对于长查询（耗时超过 500 毫秒的查询），慢启动可能会增加延迟，但对于短查询，速度更快。默认值为 10 毫秒。

citus.max_cached_conns_per_worker（整数）

每个后端都会打开与Worker节点的连接以查询分片。在事务结束时，配置的连接数保持打开状态以加速后续命令。增加此值将减少多分片查询的延迟，但也会增加Worker节点的开销。

默认值为 1。较大的值（例如 2）可能对使用少量并发会话的集群有所帮助，但走得太远是不明智的（例如，16 会太高）。

citus.force_max_query_parallelization（布尔值）

模拟已弃用的real-time executor。这用于打开尽可能多的连接以最大化查询并行度。

启用此 GUC 后，Citus 将强制adaptive executor在执行分布式查询时使用尽可能多的连接。如果未启用，executor可能会选择使用较少的连接来优化整体查询执行吞吐量。在内部，将此设置为 true 将最终为每个任务使用一个连接。

一个有用的地方是在一个事务中，它的第一个查询是轻量级的并且需要很少的连接，而随后的查询将受益于更多的连接。 Citus 根据第一条语句决定在事务中使用多少连接，除非我们使用 GUC 提供提示，否则它可以限制其他查询。

BEGIN;

add this hint

SET citus.force_max_query_parallelization TO ON;

a lightweight query that doesn't require many connections

SELECT count(*) FROM table WHERE filter = x;

a query that benefits from more connections, and can obtain

them since we forced max parallelization above

SELECT ... very .. complex .. SQL;

COMMIT;

默认值为false.

Task-tracker executor

citus.task_tracker_delay（整数）

此参数设置每完成一个任务管理轮次后task-tracker的睡眠时间，默认值为 200 毫秒。 task-tracker进程定期唤醒，逐个查看分配给它的所有任务，然后计划并执行这些任务。然后，task-tracker将休眠一段时间，唤醒后再次逐个查看这些任务。此配置值确定睡眠时长。此参数在Worker节点上生效，需要在 spacture 文件中进行更改。编辑配置文件后，用户可以发送 SIGHUP 信号或重启服务器，使更改生效。

可以减小此参数，以通过减少管理轮次之间的时间差来缩减task-tracker executor造成的延迟。如果分片查询较短，因而会定期更新其状态，则降低延迟会很有作用。

citus.max_tracked_tasks_per_node（整数）

设置每个节点的最大跟踪任务数，默认为 1024。此配置值限制了用于跟踪分配任务的哈希表的大小，因此限制了在任何给定时间可以跟踪的最大任务数。该值只能在服务器启动时设置，对Worker节点有效。

如果希望每个Worker节点能够跟踪更多任务，则需要增加此参数。如果此值低于所需值，Citus 会在Worker节点上出错，指出它的共享内存不足，并提示增加此参数可能会有所帮助。

citus.max_assign_task_batch_size （整数）

Coordinator节点上的task-tracker以同步方式将任务分批分配到Worker节点上的守护程序。此参数设置可在一个批中分配的最大任务数。选择较大的批大小可以加快任务分配。但是，如果Worker节点数量很大，则所有的Worker节点可能需要花费更长的时间来获取任务。此参数可在运行时设置，并在Coordinator节点上生效。

citus.max_running_tasks_per_node（整数）

task-tracker进程将会适当地计划并执行分配给它的任务。此配置值设置在任意给定时间，在一个节点上可并发执行的最大任务数，默认值为 8。该参数对Worker节点有效，需要在 postgresql.conf 文件中更改。编辑完配置文件后，用户可以发送 SIGHUP 信号或重新启动服务器以使更改生效。

该限制可确保不会有许多任务同时命中磁盘，并有助于避免磁盘 I/O 争用。如果从内存或 SSD 为查询提供服务，则你可以放心地增大 max_running_tasks_per_node。

citus.partition_buffer_size（整数）

设置用于分区操作的缓冲区大小，默认值为 8 MB。 Citus 允许在连接两个大型表时将表数据重新分区到多个文件中。填满此分区缓冲区后，已重新分区的数据将刷新到磁盘上的文件中。此配置可在运行时设置，并在Worker节点上生效。

max_connections（整数）

设置到数据库服务器的最大并发连接数。默认值通常为 100 个连接，但如果cpu 设置不支持它（在 initdb 期间确定），则可能会更少。realtime executor为其发送查询的每个分片维护一个连接。增加这个配置参数将允许executor有更多的并发连接，从而并行处理更多的分片。此参数必须在Worker节点和Coordinator节点上更改，并且只能在服务器启动期间完成。

max_files_per_process（整数）

为每个服务器进程设置同时打开的最大文件数，默认为 1000。Realtime executor需要为每个向其发送查询的分片提供两个文件描述符。增加此配置参数将允许executor拥有更多打开的文件描述符，从而并行处理更多分片。此更改必须在Worker节点和Coordinator节点上进行，并且只能在服务器启动期间完成。

Note: 与 max_files_per_process 一起，可能还必须使用 ulimit 命令增加每个进程的打开文件描述符的cpu 限制。

Explain 输出

citus.explain_all_tasks（布尔值）

默认情况下，对分布式查询运行 EXPLAIN 时，Citus 将显示单个任意任务的输出。在大多数情况下，explain 输出在各个任务中都是类似的。偶尔会以不同的方式规划某些任务，或者某些任务的执行时间要长得多。在这种情况下，启用此参数会很有用，EXPLAIN 输出会在此参数之后包括所有任务。解释所有任务可能会导致 EXPLAIN 花费更长的时间。

常见问题（FAQ）

我可以在分布式表上创建主键吗？

目前，仅当分布列是主键的一部分时，Citus 才施加主键约束。这确保只需要在一个分片上检查约束以确保唯一性。

如何将节点添加到现有 Citus 集群？

通过使用新节点的主机名（或 IP 地址）和端口号调用 master_add_node UDF 来手动添加节点。

无论哪种方式，在将节点添加到现有集群后，它都不会包含任何数据（分片）。 Citus 将开始将任何新创建的分片分配给该节点。

Citus 如何处理Worker节点故障？

Citus 支持两种复制模式，允许它容忍Worker节点故障。在第一个模型中，我们使用 Spacture 的流式复制来按原样复制整个Worker节点。在第二种模型中，Citus 可以复制数据修改语句，从而跨不同的Worker节点复制分片。它们具有不同的优势，具体取决于工作负载和使用场景，如下所述：

Spacture 流式复制。此选项最适合繁重的 OLTP 工作负载。它通过连续将 WAL 记录流式传输到备用节点来复制整个Worker节点。
Citus 分片复制。此选项最适合仅append 工作负载。 Citus 通过自动复制 DML 语句和管理一致性来跨不同节点复制分片。如果一个节点出现故障，Coordinator节点将通过无缝地将工作路由到副本来继续为查询提供服务。要启用分片复制，只需设置 SET citus.shard_replication_factor = 2; （或更高）在将数据分发到集群之前。

Citus 如何处理Coordinator节点的故障转移？

由于 Citus Coordinator节点类似于标准的 Spacture 服务器，因此可以使用常规的 Spacture 同步复制和故障转移来提供Coordinator节点的更高可用性。以这种方式使用同步复制来增加对Coordinator节点故障的恢复能力。

如何将查询结果提取到分布式表中？

当表满足Co-Location时，Citus 支持 INSERT / SELECT 语法，用于将分布式表上的查询结果复制到分布式表中。

如果表不是Co-Location，或者正在使用append 分发，则可以使用一些变通方法（例如，使用 COPY 将数据复制出来，然后再复制回目标表）。

我可以在同一个查询中将分布式表和非分布式表连接在一起吗？

如果要在小维度表（常规 Postgres 表）和大表（分布式）之间进行连接，则将本地表包装在子查询中。 Citus 的子查询执行逻辑将允许连接工作。

是否有任何 Citus 不支持的 Spacture 功能？

由于 Citus 通过扩展 Spacture 提供分布式功能，因此它使用标准的 Spacture SQL 结构。支持绝大多数查询，即使它们在网络上合并来自多个数据库节点的数据时也是如此。这包括跨节点的事务语义。目前支持所有 SQL，除了：

相关子查询 (Correlated subqueries)
递归CTE (Recursive CTEs)
表样本 (Table sample)
选择……更新 (SELECT … FOR UPDATE)
分组集 (Grouping sets)
PARTITION BY 中不包含分布列的窗口函数

此外，Citus 对访问数据库集群中单个节点的查询提供 100% 的 SQL 支持。这些查询很常见，例如，在不同节点存储不同租户的多租户应用程序中。

请记住，即使有如此广泛的 SQL 覆盖范围，数据建模也会对查询性能产生重大影响。

对数据进行哈希分区时，如何选择分片数？

首次分配表时的选择之一是其分片数。可以为每个表设置不同的设置，最佳值取决于使用场景。创建集群后更改shard数是可能的，但很困难，因此请使用这些准则来选择正确的大小。

在多租户数据库使用场景中，我们建议在 32 - 128 个分片之间进行选择。对于小于 100GB 的较小工作负载，可以从 32 个分片开始，对于较大的工作负载，可以选择 64 或 128 个。这意味着有从 32 扩展到 128 个工作机器的余地。

在实时分析使用场景中，分片数应与Worker节点上的cpu 总数相关。为确保最大并行度，应该在每个节点上创建足够的分片，以便每个 CPU至少有一个分片。我们通常建议创建大量的初始分片，例如当前 CPU 数量的 2 倍或 4 倍。如果添加更多Worker节点和 CPU ，这将允许未来扩展。

要为要分发的表选择分片数，请更新 citus.shard_count 变量。这会影响对 create_distributed_table 的后续调用。例如

SET citus.shard_count = 64;

any tables distributed at this point will have

sixty-four shards

citus 如何支持计数（不同）查询？

Citus 可以评估Worker节点内和跨Worker节点的count(distinct)聚合。在表的分布列上聚合时，Citus 可以在Worker节点内向下推计数并汇总结果。否则它可以将不同的行拉到Coordinator节点并在那里计算。如果将数据传输到Coordinator节点的成本太高，也可以使用快速近似计数。

分布式表在哪些情况下支持唯一性约束？

只有当受约束的列包含分布列时，Citus 才能强制执行主键或唯一性约束。特别是这意味着如果单个列构成主键，那么它也必须是分布列。

此限制允许 Citus 将唯一性检查本地化到单个分片，并让Worker节点上的 Spacture 有效地进行检查。

如何在 Citus 集群中创建数据库角色、功能、扩展等？

某些命令在Coordinator节点上运行时，不会分发到Worker节点：

创建角色/用户 (CREATE ROLE/USER)
创建函数 (CREATE FUNCTION)
创建类型 (CREATE TYPE)
创建扩展 (CREATE EXTENSION)
创建数据库 (CREATE DATABASE)
ALTER … SET SCHEMA
ALTER TABLE ALL IN TABLESPACE

仍然可以通过在所有节点上显式运行这些命令来使用它们。 Citus 提供了一个功能来在所有Worker节点之间执行查询：

SELECT run_command_on_workers($cmd$
  /* the command to run */
  CREATE FUNCTION ...
$cmd$);

在手动查询传播中了解更多信息。另请注意，即使在手动传播 CREATE DATABASE 之后，必须执行create extension citus来安装citus。

如果Worker节点的地址发生变化怎么办？

如果 Worker 的主机名或 IP 地址发生变化，需要使用 master_update_node 让Coordinator节点知道：

update worker node metadata on the coordinator
(remember to replace 'old-address' and 'new-address'
with the actual values for your situation)
select master_update_node(nodeid, 'new-address', nodeport)
  from pg_dist_node
 where nodename = 'old-address';

在执行此更新之前，Coordinator节点将无法与该Worker节点通信以进行查询。

哪个分片包含特定租户的数据？

Citus 提供 UDF 和元数据表来确定分布列值到特定分片的映射，以及分片在Worker节点上的分布。

我忘记了一个表的分布列，我如何找到它？

Citus Coordinator节点元数据表包含此信息。

我可以通过多个键分配一个表吗？

不可以，必须为每个表选择一列作为分布列。人们希望按两列分布的常见场景是时间序列数据。但是，对于这种情况，我们建议在非时间列上使用哈希分布，并将其与时间列上的 Spacture 分区相结合，如时间序列数据中所述。

为什么 pg_relation_size 报告分布式表的零字节？

分布式表中的数据存在于Worker节点（在分片中），而不是Coordinator节点上。分布式表大小的真实度量是通过分片大小的总和获得的。 Citus 提供了辅助函数来查询这些信息。

为什么我会看到有关 max_intermediate_result_size 的错误？

Citus 必须使用多个步骤来运行一些具有子查询或 CTE 的查询。使用 Subquery/CTE Push-Pull Execution，它将子查询结果推送到所有Worker节点以供主查询使用。如果这些结果太大，这可能会导致无法接受的网络开销，甚至可能导致累积和分发结果的Coordinator节点上的存储空间不足。

Citus 有一个可配置的设置，citus.max_intermediate_result_size 来指定一个子查询结果大小阈值，在该阈值处查询将被取消。如果遇到错误，它看起来像：

ERROR:  the intermediate result size exceeds
citus.max_intermediate_result_size (currently 1 GB)

DETAIL:  Citus restricts the size of intermediate results of complex
subqueries **and** CTEs to avoid accidentally pulling large result sets
into once place.

HINT:  To run the current query, set citus.max_intermediate_result_size
to a higher value **or** -1 to disable.

正如错误消息所暗示的那样，可以（谨慎地）通过更改变量来增加此限制:

SET citus.max_intermediate_result_size = '3GB';

我可以在 Citus 上为多租户应用程序按模式分片吗？

事实证明，虽然在与租户打交道时将每个租户的信息存储在单独的模式中可能是一种有吸引力的开始方式，但它会导致问题。在 Citus 中，我们按租户 ID 进行分区，一个分片可以包含来自多个租户的数据。

快速入门

Citus简介

Citus适用场景

实践示例

Multi-tenant Applications多租户应用程序

关键概念

Nodes

Coordinator节点 和 Workers节点

Distributed Data

Shards

Co-Location

Parallelism

SQL参考文档

DDL

DML

使用汇总缓存聚合

查询分布式表（SQL）

查询处理

手动查询传播

SQL 支持和解决方法

Citus API

Citus 函数

Citus 表和视图 (Citus Tables and Views)

配置参考 (Configuration Reference)

常见问题（FAQ）

Coordinator节点和 Workers节点