Citus 分布式 PostgreSQL 集群-SQL Reference(查詢處理)

作者：為少 2022-03-30 19:18:31

對(duì)于 SELECT 查詢，計(jì)劃器首先創(chuàng)建輸入查詢的計(jì)劃樹，并將其轉(zhuǎn)換為可交換和關(guān)聯(lián)形式，以便可以并行化。它還應(yīng)用了一些優(yōu)化以確保以可擴(kuò)展的方式執(zhí)行查詢，并最大限度地減少網(wǎng)絡(luò) I/O。

一個(gè) Citus 集群由一個(gè) coordinator 實(shí)例和多個(gè) worker 實(shí)例組成。數(shù)據(jù)在 worker 上進(jìn)行分片和復(fù)制，而 coordinator 存儲(chǔ)有關(guān)這些分片的元數(shù)據(jù)。向集群發(fā)出的所有查詢都通過(guò) coordinator 執(zhí)行。 coordinator 將查詢劃分為更小的查詢片段，其中每個(gè)查詢片段可以在分片上獨(dú)立運(yùn)行。然后協(xié)調(diào)器將查詢片段分配給 worker，監(jiān)督他們的執(zhí)行，合并他們的結(jié)果，并將最終結(jié)果返回給用戶。查詢處理架構(gòu)可以通過(guò)下圖進(jìn)行簡(jiǎn)要描述。

Citus 的查詢處理管道涉及兩個(gè)組件：

分布式查詢計(jì)劃器和執(zhí)行器
PostgreSQL 計(jì)劃器和執(zhí)行器

我們將在后續(xù)部分中更詳細(xì)地討論它們。

分布式查詢計(jì)劃器

Citus 的分布式查詢計(jì)劃器接收 SQL 查詢并規(guī)劃它以進(jìn)行分布式執(zhí)行。

接下來(lái)，計(jì)劃器將查詢分為兩部分 - 在 coordinator 上運(yùn)行的 coordinator 查詢和在 worker 上的各個(gè)分片上運(yùn)行的 worker 查詢片段。然后，計(jì)劃器將這些查詢片段分配給 worker，以便有效地使用他們的所有資源。在這一步之后，分布式查詢計(jì)劃被傳遞給分布式執(zhí)行器執(zhí)行。

分布列上的鍵值查找或修改查詢的規(guī)劃過(guò)程略有不同，因?yàn)樗鼈兦『妹幸粋€(gè)分片。一旦計(jì)劃器收到傳入的查詢，它需要決定查詢應(yīng)該路由到的正確分片。為此，它提取傳入行中的分布列并查找元數(shù)據(jù)以確定查詢的正確分片。然后，計(jì)劃器重寫該命令的 SQL 以引用分片表而不是原始表。然后將該重寫的計(jì)劃傳遞給分布式執(zhí)行器。

分布式查詢執(zhí)行器

Citus 的分布式執(zhí)行器運(yùn)行分布式查詢計(jì)劃并處理故障。執(zhí)行器非常適合快速響應(yīng)涉及過(guò)濾器、聚合和共置連接的查詢，以及運(yùn)行具有完整 SQL 覆蓋的單租戶查詢。它根據(jù)需要為每個(gè)分片打開(kāi)一個(gè)與 woker 的連接，并將所有片段查詢發(fā)送給他們。然后它從每個(gè)片段查詢中獲取結(jié)果，合并它們，并將最終結(jié)果返回給用戶。

子查詢/CTE Push-Pull 執(zhí)行

如有必要，Citus 可以將來(lái)自子查詢和 CTE 的結(jié)果收集到 coordinator 節(jié)點(diǎn)中，然后將它們推送回 worker 以供外部查詢使用。這允許 Citus 支持更多種類的 SQL 構(gòu)造。

例如，在 WHERE 子句中包含子查詢有時(shí)不能與主查詢同時(shí)執(zhí)行內(nèi)聯(lián)，而必須單獨(dú)執(zhí)行。假設(shè) Web 分析應(yīng)用程序維護(hù)一個(gè)按 page_id 分區(qū)的 page_views 表。要查詢前 20 個(gè)訪問(wèn)量最大的頁(yè)面上的訪問(wèn)者主機(jī)數(shù)，我們可以使用子查詢來(lái)查找頁(yè)面列表，然后使用外部查詢來(lái)計(jì)算主機(jī)數(shù)。

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

執(zhí)行器希望通過(guò) page_id 對(duì)每個(gè)分片運(yùn)行此查詢的片段，計(jì)算不同的 host_ips，并在 coordinator 上組合結(jié)果。但是，子查詢中的 LIMIT 意味著子查詢不能作為片段的一部分執(zhí)行。通過(guò)遞歸規(guī)劃查詢，Citus 可以單獨(dú)運(yùn)行子查詢，將結(jié)果推送給所有 worker，運(yùn)行主片段查詢，并將結(jié)果拉回 coordinator。 push-pull(推拉) 設(shè)計(jì)支持上述子查詢。

讓我們通過(guò)查看此查詢的 EXPLAIN 輸出來(lái)了解這一點(diǎn)。它相當(dāng)參與：

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 Adaptive)  (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 Adaptive)  (cost=0.00..0.00 rows=0 width=0)
                Task Count: 32
                Tasks Shown: One of 32
                ->  Task
                  Node: host=localhost port=9701 dbname=postgres
                  ->  HashAggregate  (cost=54.70..56.70 rows=200 width=12)
                    Group Key: page_id
                    ->  Seq Scan on page_views_102008 page_views  (cost=0.00..43.47 rows=2247 width=4)
      Task Count: 32
      Tasks Shown: One of 32
      ->  Task
        Node: host=localhost port=9701 dbname=postgres
        ->  HashAggregate  (cost=84.50..86.75 rows=225 width=36)
          Group Key: page_views.page_id, page_views.host_ip
          ->  Hash Join  (cost=17.00..78.88 rows=1124 width=36)
            Hash Cond: (page_views.page_id = intermediate_result.page_id)
            ->  Seq Scan on page_views_102008 page_views  (cost=0.00..43.47 rows=2247 width=36)
            ->  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)

讓我們把它拆開(kāi)并檢查每一塊。

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

樹的 root 是 coordinator 節(jié)點(diǎn)對(duì) worker 的結(jié)果所做的事情。在這種情況下，它正在對(duì)它們進(jìn)行分組，并且 GroupAggregate 要求首先對(duì)它們進(jìn)行排序。

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

自定義掃描有兩個(gè)大子樹，從“分布式子計(jì)劃”開(kāi)始。

        ->  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 Adaptive)  (cost=0.00..0.00 rows=0 width=0)
                Task Count: 32
                Tasks Shown: One of 32
                ->  Task
                  Node: host=localhost port=9701 dbname=postgres
                  ->  HashAggregate  (cost=54.70..56.70 rows=200 width=12)
                    Group Key: page_id
                    ->  Seq Scan on page_views_102008 page_views  (cost=0.00..43.47 rows=2247 width=4)
.

工作節(jié)點(diǎn)為 32 個(gè)分片中的每一個(gè)運(yùn)行上述內(nèi)容(Citus 正在選擇一個(gè)代表進(jìn)行顯示)。我們可以識(shí)別 IN (...) 子查詢的所有部分：排序、分組和限制。當(dāng)所有 worker 完成此查詢后，他們會(huì)將其輸出發(fā)送回 coordinator，coordinator 將其組合為“中間結(jié)果”。

      Task Count: 32
      Tasks Shown: One of 32
      ->  Task
        Node: host=localhost port=9701 dbname=postgres
        ->  HashAggregate  (cost=84.50..86.75 rows=225 width=36)
          Group Key: page_views.page_id, page_views.host_ip
          ->  Hash Join  (cost=17.00..78.88 rows=1124 width=36)
            Hash Cond: (page_views.page_id = intermediate_result.page_id)
.

Citus 在第二個(gè)子樹中啟動(dòng)另一個(gè)執(zhí)行器作業(yè)。它將在 page_views 中計(jì)算不同的主機(jī)。它使用 JOIN 連接中間結(jié)果。中間結(jié)果將幫助它限制在前二十頁(yè)。

       ->  Seq Scan on page_views_102008 page_views  (cost=0.00..43.47 rows=2247 width=36)
            ->  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)
.

工作人員使用 read_intermediate_result 函數(shù)在內(nèi)部檢索中間結(jié)果，該函數(shù)從 coordinator 節(jié)點(diǎn)復(fù)制的文件中加載數(shù)據(jù)。

這個(gè)例子展示了 Citus 如何使用分布式子計(jì)劃在多個(gè)步驟中執(zhí)行查詢，以及如何使用 EXPLAIN 來(lái)了解分布式查詢執(zhí)行。

PostgreSQL 計(jì)劃器和執(zhí)行器

一旦分布式執(zhí)行器將查詢片段發(fā)送給 worker，它們就會(huì)像常規(guī) PostgreSQL 查詢一樣被處理。該 worker 上的 PostgreSQL 計(jì)劃程序選擇在相應(yīng)分片表上本地執(zhí)行該查詢的最佳計(jì)劃。 PostgreSQL 執(zhí)行器然后運(yùn)行該查詢并將查詢結(jié)果返回給分布式執(zhí)行器。您可以從 PostgreSQL 手冊(cè)中了解有關(guān) PostgreSQL 計(jì)劃器和執(zhí)行器的更多信息。最后，分布式執(zhí)行器將結(jié)果傳遞給 coordinator 進(jìn)行最終聚合。