tuhaihe commented on code in PR #1971: URL: https://github.com/apache/cloudberry/pull/1971#discussion_r3985262760
########## docs/core-internals/part-1-introduction/ch01.md: ########## @@ -0,0 +1,711 @@ +--- +id: ch01 +title: "1. Introduction: The Big Picture" +sidebar_position: 1 +--- + +<!-- + Licensed to the Apache Software Foundation (ASF) under one + or more contributor license agreements. See the NOTICE file + distributed with this work for additional information + regarding copyright ownership. The ASF licenses this file + to you under the Apache License, Version 2.0 (the + "License"); you may not use this file except in compliance + with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, + software distributed under the License is distributed on an + "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + KIND, either express or implied. See the License for the + specific language governing permissions and limitations + under the License. +--> + +# 1. Introduction: The Big Picture + +```mermaid +flowchart TB + CL["Client"]:::ext + subgraph QD["QD — coordinator · Gp_role = GP_ROLE_DISPATCH"] + direction LR + PL["Parser → Planner / ORCA<br/>cuts the plan into slices at each Motion · §15–§17"]:::q + DP["Dispatcher — CdbDispatchPlan<br/>allocates gangs, ships plan + snapshot, coordinates 2PC · §13, §12"]:::q + PL --> DP + end + IC["INTERCONNECT — UDPIFC · §19<br/>Motion between gangs, and the final Gather to the QD"]:::net + subgraph SEG0["Segment 0 — forks one QE per slice, all sharing one snapshot"] + direction TB + Q0a["QE · Executor · §18"]:::q + Q0b["QE · Executor"]:::q + SS0["SharedLocalSnapshot<br/>one consistent view for the segment's QEs"]:::t + D0[("data shard 0")]:::s + Q0a --- SS0 + Q0b --- SS0 + Q0a -->|"local scan / write"| D0 + Q0b -->|"local"| D0 + end + subgraph SEG1["Segment 1 — its own QEs + shard"] + direction TB + Q1a["QE · Executor"]:::q + Q1b["QE · Executor"]:::q + SS1["SharedLocalSnapshot"]:::t + D1[("data shard 1")]:::s + Q1a --- SS1 + Q1b --- SS1 + Q1a -->|"local"| D1 + Q1b -->|"local"| D1 + end + LK["Locks · §11<br/>each QE locks the objects it touches"]:::t + WALN["WAL · §20<br/>each segment logs its own changes"]:::h + CL ==>|"SQL"| PL + DP ==>|"dispatch plan + snapshot — over libpq"| SEG0 + DP ==>|"libpq"| SEG1 + Q0a -. Motion .- IC + Q1a -. Motion .- IC + IC ==>|"Gather results"| DP + DP ==>|"rows"| CL + LK -.-> SEG0 + LK -.-> SEG1 + SEG0 -.->|"changes logged"| WALN + SEG1 -.->|"changes logged"| WALN + classDef ext fill:#ffffff,stroke:#1f2328,color:#1f2328 + classDef q fill:#ddf4ff,stroke:#0969da,color:#0a3069 + classDef s fill:#dafbe1,stroke:#1a7f37,color:#04260f + classDef net fill:#fff8c5,stroke:#bf8700,stroke-width:2px,color:#3b2300 + classDef t fill:#fff1e5,stroke:#bc4c00,color:#5e2700 + classDef h fill:#ffebe9,stroke:#cf222e,color:#5c0011 + style QD fill:#f3f8ff,stroke:#0969da + style SEG0 fill:#f3fbf5,stroke:#1a7f37 + style SEG1 fill:#f3fbf5,stroke:#1a7f37 +``` + +*How a query really runs. On the coordinator the single **QD** (`Gp_role = GP_ROLE_DISPATCH`) parses and plans the query, cutting it into **slices** at each Motion; its **Dispatcher** (`CdbDispatchPlan`) allocates **gangs** of QEs and ships the plan with the **distributed snapshot** to them over **libpq**, then coordinates the 2PC commit (§13, §12). A segment may **fork several QEs** — one per slice — which share one **SharedLocalSnapshot** so they see the same data; each **Executor** scans and writes its segment's shard *locally* (§18), takes the **locks** it needs (§11), and logs its changes to that segment's **WAL** (§20). Reaching another segment's data — and the final **Gather** to the QD — travels over the **Interconnect** (UDPIFC) as **Motion**, a path separate from the libpq dispatch (§19).* + +## 1.1 The MPP Cluster: Coordinator, Segments & Distribution + +**本节要点:** `shared-nothing` · `distribution policies` · `cdbhash → segment` · `cluster catalogs` + +```mermaid +flowchart TB + CL["Client — SQL"]:::ext + M["Master / Coordinator<br/>Catalog (metadata) · no user data"]:::q + subgraph SEGS["Segments — shared-nothing; the user data lives here"] + direction LR + S0["Segment 0<br/>Catalog + data shard"]:::s + S1["Segment 1<br/>Catalog + data shard"]:::s + end + CL ==>|"connect"| M + M ---|"Interconnect"| SEGS + ROWS["table rows · DISTRIBUTED BY (key)"]:::ext -.->|"cdbhash(key) → one segment"| SEGS + classDef ext fill:#ffffff,stroke:#1f2328,color:#1f2328 + classDef q fill:#ddf4ff,stroke:#0969da,color:#0a3069 + classDef s fill:#dafbe1,stroke:#1a7f37,color:#04260f + style SEGS fill:#f3fbf5,stroke:#1a7f37 +``` + +*The cluster, in its simplest form. One **Master** holds the catalog (system metadata) but no user data; the **Segments** each hold the catalog *and* a shard of the user data — so the catalog is replicated everywhere, while the data is sharded. A table's rows are spread across the segments by a hash of its distribution key: `cdbhash(key)` deterministically sends each row to one segment ([§1.1](#11-the-mpp-cluster-coordinator-segments--distribution).2–[§1.1](#11-the-mpp-cluster-coordinator-segments--distribution).3). Master and Segments communicate over the Interconnect (§19).* + +### 1.1.1 A shared-nothing cluster + +Cloudberry is one logical database spread across many PostgreSQL processes that share *nothing* — no shared memory, no shared disk. One process is the **coordinator**: it accepts connections, parses and plans queries, dispatches work, and gathers results, but it stores **no user data**. The rest are **segments**, each a full PostgreSQL instance with its own CPU, memory, and disk, holding a slice of every distributed table and executing the plan on its slice in parallel. Scaling out means adding segments. + +For availability each primary segment can have a **mirror** — a streaming-replication copy on another host ready to take over (Chapter 20). The whole cluster is described by one catalog, `gp_segment_configuration`, readable from the coordinator: + +*The cluster map. `content` −1 is the coordinator; `content` 0 and up are segments. This is a demo cluster — one coordinator and three primary segments, no mirror.* + +```sql +SELECT content, role, preferred_role, mode, status, port +FROM gp_segment_configuration ORDER BY content; +``` + +```text {3,4} title="output" + content | role | preferred_role | mode | status | port +---------+------+----------------+------+--------+------ + -1 | p | p | n | u | 7100 + 0 | p | p | n | u | 7102 + 1 | p | p | n | u | 7103 + 2 | p | p | n | u | 7104 +(4 rows) +``` + +- **content** — the segment's logical id: **−1** is the coordinator, **0 … N−1** are the data segments. A primary and its mirror share one content. +- **role / preferred_role** — `p` primary or `m` mirror — current vs. the role it owns when healthy; they differ after a failover. +- **mode / status** — sync state (`s` in-sync, `n` not-replicating) and liveness (`u` up, `d` down), maintained by the fault-tolerance service (Chapter 20). + +:::note + +A production cluster has many segments, each with a mirror on a different host. Everything below is the same with more rows in this table — this demo has three segments, so every distributed table is sharded across `content` 0, 1, and 2. + +::: + +### 1.1.2 Three distribution policies + +Every table carries a **distribution policy** that decides how its rows are spread across the segments. There are three, chosen at `CREATE TABLE`: + +- **`DISTRIBUTED BY (cols)`** — **Hashed.** Each row's key columns are hashed to pick its segment. The default, and what you want for big tables: even spread, and joins on the key need no data movement. +- **`DISTRIBUTED REPLICATED`** — **Replicated.** A full copy of the table on *every* segment. Good for small dimension tables, so joins against them never have to redistribute. +- **`DISTRIBUTED RANDOMLY`** — **Random.** Rows go round-robin to segments with no key. Even spread, but every join on it requires a redistribution Motion (Chapter 19). + +The policy is recorded per table in `gp_distribution_policy` — `policytype` is `p` (partitioned: hashed or random) or `r` (replicated); `distkey` lists the key column numbers, and is empty for random and replicated tables: + +*The three policies side by side. `c1dist` hashes on column 1; the replicated copy is `r`; the random table is `p` with an empty `distkey`.* + +```sql +CREATE TABLE c1dist(id int, payload text) DISTRIBUTED BY (id); +CREATE TABLE c1dist_rep(id int) DISTRIBUTED REPLICATED; +CREATE TABLE c1dist_rand(id int) DISTRIBUTED RANDOMLY; + +SELECT localoid::regclass AS relname, policytype, distkey +FROM gp_distribution_policy +WHERE localoid IN ('c1dist'::regclass,'c1dist_rep'::regclass,'c1dist_rand'::regclass) +ORDER BY relname; +``` + +```text {3,4,5} title="output" + relname | policytype | distkey +-------------+------------+--------- + c1dist | p | 1 + c1dist_rep | r | + c1dist_rand | p | +(3 rows) +``` + +:::note + +The enum behind `policytype` is `POLICYTYPE_PARTITIONED` (`p`) vs `POLICYTYPE_REPLICATED` (`r`) — `src/include/catalog/gp_distribution_policy.h:88`; the catalog row is `policytype` + `distkey` at `gp_distribution_policy.h:33`. A third value, `POLICYTYPE_ENTRY`, marks coordinator-only (entry) tables such as some catalogs. + +::: + +### 1.1.3 How a row finds its segment + +For a hash-distributed table, the segment that owns a row is computed, not looked up. The key columns are fed through `cdbhash` to build a 32-bit hash; `cdbhashreduce` then folds that hash down to a segment number: + +`row's distribution key` → `cdbhash(key) → 32-bit hash` → `cdbhashreduce → segment id` → `route to that segment` + +```c title="src/backend/cdb/cdbhash.c:253" +/* cdbhashreduce: map the 32-bit hash down to one segment */ +case REDUCE_LAZYMOD: + result = (h->hash) % (h->numsegs); /* hash mod number-of-segments */ +``` + +Because the mapping is pure arithmetic on the key, every process — coordinator planner and all segments — agrees on where a row lives without coordination. Insert 100 000 rows and ask each segment how many it holds; on this three-segment cluster `cdbhash` spreads them roughly evenly: + +*Per-segment row counts via the `gp_segment_id` system column. With three segments, `cdbhash` spreads the 100 000 rows roughly evenly across content 0, 1, and 2.* + +```sql +INSERT INTO c1dist SELECT g, 'x' FROM generate_series(1,100000) g; +SELECT gp_segment_id, count(*) FROM c1dist GROUP BY gp_segment_id ORDER BY gp_segment_id; +``` + +```text {3} title="output" + gp_segment_id | count +---------------+-------- + 0 | 33462 + 1 | 33327 + 2 | 33211 +(3 rows) +``` + +:::note + +`cdbhash` lives at `src/backend/cdb/cdbhash.c:189` and `cdbhashreduce` at `:253` (with bitmask, lazy-mod, and jump-hash reducers); a keyless random table instead calls `cdbhashrandomseg` (`:291`). The hash is what lets two tables distributed on the same key be joined segment-local, with no data movement. + +::: + +### 1.1.4 The cluster catalogs + +Three small catalogs tie the model together — the map, the per-table policy, and each process's sense of identity: + +- **`gp_segment_configuration`** — the cluster map: one row per segment (and the coordinator), with role, mode, status, host, and port. A shared catalog — `src/include/catalog/gp_segment_configuration.h:44`. +- **`gp_distribution_policy`** — one row per distributed table: its `policytype` and `distkey` ([§1.1](#11-the-mpp-cluster-coordinator-segments--distribution).2) — `gp_distribution_policy.h:30`. +- **`GpIdentity` / `gp_id`** — every process knows its own `content` id. The coordinator's is `MASTER_CONTENT_ID` (−1), which is exactly how `IS_QUERY_DISPATCHER()` is defined — `src/include/cdb/cdbvars.h:765,774`. The `gp_segment_id` column we just grouped by is this identity surfaced per row. + +:::note + +These three answer the cluster's basic questions: **where are the segments** (gp_segment_configuration), **how is this table spread** (gp_distribution_policy), and **who am I** (GpIdentity). Everything in later chapters — planning Motions, dispatching gangs, recovering a failed segment — reads from them. + +::: + +## 1.2 Data Organization + +**本节要点:** `databases · tablespaces` · `files & forks` · `pages & TOAST` + +```mermaid +flowchart TB + subgraph L["Logical — what SQL sees"] + direction TB + DB["database"] --> SCH["schema"] --> REL["relation — a table or index"] + end + subgraph P["Physical — files on each segment's disk"] + direction TB + FORKS["relation = a set of forks"] --> MAIN["main fork<br/>row or index data"] + FORKS --> FSM["fsm fork<br/>free-space map"] + FORKS --> VM["vm fork<br/>visibility map"] + MAIN --> PAGES["fixed 32 KB pages"] + end + REL -->|"stored as"| FORKS + TS["tablespace = a directory"] -.->|"holds the files"| P +``` + +*From SQL down to bytes. A database holds schemas, which hold relations; each relation is physically a set of forks (main, fsm, vm) made of fixed 32 KB pages, living in a tablespace directory. In an MPP cluster this whole physical picture exists independently on every segment ([§1.1](#11-the-mpp-cluster-coordinator-segments--distribution)).* + +Before any query runs it helps to know how Cloudberry lays data out — from the database a client connects to, down to the files on a segment's disk. This section walks that path top to bottom. + +### 1.2.1 Databases, schemas, and tablespaces + +A running Apache Cloudberry system — a *cluster* in PostgreSQL terms — serves a set of **databases**. A client connects to exactly one; objects in other databases are not directly visible. Inside a database, **schemas** are namespaces that group tables, indexes, views, and functions (every database starts with a `public` schema and the system's `pg_catalog`). A fresh cluster carries the familiar three databases — your working `postgres` and the two `template` databases that `initdb` creates: + +*The databases and tablespaces of a fresh cluster.* + +```sql +SELECT datname FROM pg_database ORDER BY 1; +SELECT spcname FROM pg_tablespace ORDER BY 1; +``` + +```text title="output" + datname +----------- + postgres + template0 + template1 +(3 rows) + + spcname +------------ + pg_default + pg_global +(2 rows) +``` + +Where schemas organize objects *logically*, a **tablespace** decides *where on disk* their files go: it is essentially a named directory. The two built-ins are `pg_default` (the data directory's `base/`) and `pg_global` (shared catalogs). Creating a tablespace on a faster disk and placing hot tables there is a physical-placement decision that does not change a single SQL name. + +And `pg_default` or `pg_global` is the same *name* on every host, but the *directory* it resolves to is per-segment: each segment is a separate instance with its own data directory, beneath which a relation lives at the same relative path. `gp_segment_configuration` shows those roots: + +*Two built-in tablespaces, and each instance's own data directory — the coordinator (content −1) and segment 0 sit under different roots, identical relative paths beneath.* + +```sql +SELECT oid, spcname FROM pg_tablespace ORDER BY oid; +SELECT role, content, datadir FROM gp_segment_configuration ORDER BY content; +``` + +```text {9,10} title="output" + oid | spcname +------+------------ + 1663 | pg_default + 1664 | pg_global +(2 rows) + + role | content | datadir +------+---------+------------------------------------------------ + p | -1 | …/gpdemo/datadirs/qddir/demoDataDir-1 + p | 0 | …/gpdemo/datadirs/dbfast1/demoDataDir0 + p | 1 | …/gpdemo/datadirs/dbfast2/demoDataDir1 + p | 2 | …/gpdemo/datadirs/dbfast3/demoDataDir2 +(4 rows) +``` + +:::note + +**The MPP twist.** Each of these names is cluster-wide, but the *storage* behind them is not centralized: every segment is itself a full PostgreSQL instance with its own copy of these databases, schemas, and tablespace directories, holding its own shard of the data. The coordinator coordinates; the segments store. That division is the whole of [§1.1](#11-the-mpp-cluster-coordinator-segments--distribution). + +::: + +### 1.2.2 Relations, files, and forks + +PostgreSQL — and so Cloudberry — calls a table or index a **relation**, and a relation is *not* a single file. Each relation is identified on disk by a triple — its tablespace and database OIDs plus its own `relfilenode` (a `RelFileNumber`). That triple is the `RelFileLocator`: + +```c title="src/include/storage/relfilelocator.h:57" +typedef struct RelFileLocator +{ + Oid spcOid; /* tablespace */ + Oid dbOid; /* database */ + RelFileNumber relNumber; /* relation (the relfilenode) */ +} RelFileLocator; +``` + +`GetRelationPath` (`src/common/relpath.c:150`) turns that triple into a path under the data directory. `pg_relation_filepath` exposes it — here, the table sits in the default tablespace (`base/`), under the database's OID, in a file named by its relfilenode: + +*A relation's on-disk path: base / database-OID / relfilenode.* + +```sql +SELECT pg_relation_filepath('c1org'); +``` + +```text {3} title="output" + pg_relation_filepath +---------------------- + base/5/16479 +(1 row) +``` + +But that one file is only the relation's **main fork**. A relation is stored as several *forks* — separate files that share the relfilenode and differ only by suffix. Their names come from a fixed table, `forkNames[]` (`src/common/relpath.c:34`), indexed by the `ForkNumber` enum (`src/include/common/relpath.h:55`): + +*The forks of a relation. The main fork has no suffix; the others append _fsm / _vm / _init.* + +| 部分 | 含义 | Review Comment: These words need to be translated into English here. FYI. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
