Citus 分布式 PostgreSQL 集群 - SQL Reference(手動查詢傳播)
手動查詢傳播
當用戶發出查詢時，Citus coordinator 將其劃分為更小的查詢片段，其中每個查詢片段可以在工作分片上獨立運行。這允許 Citus 將每個查詢分布在集群中。
但是，將查詢劃分為片段的方式(以及傳播哪些查詢)因查詢類型而異。 在某些高級情況下，手動控制此行為很有用。 Citus 提供實用函數來將 SQL 傳播到 workers、shards 或 placements。
手動查詢傳播繞過 coordinator 邏輯、鎖定和任何其他一致性檢查。 這些函數可作為最后的手段，以允許 Citus 否則不會在本機運行的語句。小心使用它們以避免數據不一致和死鎖。
在所有 Worker 上運行
最小的執行級別是廣播一條語句以在所有 worker 上執行。這對于查看整個工作數據庫的屬性很有用。
-- List the work_mem setting of each worker database
SELECT run_command_on_workers($cmd$ SHOW work_mem; $cmd$);
注意： 不應使用此命令在 worker 上創建數據庫對象，因為這樣做會使以自動方式添加 worker 節點變得更加困難。
注意： 本節中的 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$
);
在所有放置上運行
最精細的執行級別是在所有分片及其副本(也稱為放置)上運行命令。它對于運行數據修改命令很有用，這些命令必須應用于每個副本以確保一致性。
例如，假設一個分布式表有一個 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。 它將位于共置的分布式表的兩個位置的名稱插入到查詢中。放置對總是被選擇為本地的同一個 worker，其中完整的 SQL 覆蓋是可用的。因此，我們可以使用觸發器等高級 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 synchronize them so that every time little_vals
-- are created, big_vals appear with double the value
--
-- First we make a trigger function, which will
-- take the destination table placement as an argument
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;
-- 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(%L)
$cmd$
);
限制
多語句事務沒有防止死鎖的安全措施。
沒有針對中間查詢失敗和由此產生的不一致的安全措施。
查詢結果緩存在內存中; 這些函數無法處理非常大的結果集。
如果無法連接到節點，這些函數會提前出錯。
你可以做很壞的事情!
