Skip to content

PostgreSQL List 数据结构详解

摘要

PostgreSQL 的 List 是其内部广泛使用的一种数据结构,用于管理指针列表、整数列表、OID 列表和事务 ID 列表。本教程基于 PostgreSQL 源码(src/include/nodes/pg_list.hsrc/backend/nodes/list.c),详细剖析 List 的设计原理、内部实现、使用模式和最佳实践。


1. List 设计背景

1.1 历史渊源

c
/*
 * Once upon a time, parts of Postgres were written in Lisp and used real
 * cons-cell lists for major data structures. When that code was rewritten
 * in C, we initially had a faithful emulation of cons-cell lists, which
 * unsurprisingly was a performance bottleneck. A couple of major rewrites
 * later, these data structures are actually simple expansible arrays;
 * but the "List" name and a lot of the notation survives.
 */

PostgreSQL 早期使用 Lisp 风格的链表实现,后来为提升性能重写为可扩展数组实现,但仍保留了 List 的名称和部分语法约定。

1.2 核心设计原则

  1. 空列表唯一表示: 空列表始终用 NIL (NULL 指针)表示
  2. 非空列表保证长度 >= 1
  3. 列表头不会因增删元素而移动
  4. 支持四种类型: 指针列表、整数列表、OID 列表、事务 ID 列表

2. 数据结构详解

2.1 ListCell 联合体

c
// src/include/nodes/pg_list.h:45-51
typedef union ListCell
{
    void       *ptr_value;         // T_List 使用
    int         int_value;        // T_IntList 使用
    Oid         oid_value;        // T_OidList 使用
    TransactionId xid_value;      // T_XidList 使用
} ListCell;

ListCell 是一个联合体,根据列表类型存储不同类型的值。这种设计避免了类型转换的麻烦。

2.2 List 结构体

c
// src/include/nodes/pg_list.h:53-62
typedef struct List
{
    NodeTag     type;              // T_List, T_IntList, T_OidList, T_XidList
    int         length;            // 当前元素数量
    int         max_length;        // 已分配的 elements 数组长度
    ListCell   *elements;          // 可重分配的数组指针
    
    // 内联存储,避免小列表单独分配内存
    ListCell    initial_elements[FLEXIBLE_ARRAY_MEMBER];
    // 如果 elements == initial_elements,说明没有单独分配内存
} List;

2.3 内存布局图解

┌─────────────────────────────────────────────────────────────────────────┐
│                         List 内存布局                                    │
├─────────────────────────────────────────────────────────────────────────┤
│                                                                         │
│  ┌───────────────────────────────────────────────────────────────────┐  │
│  │                        List 结构体                                 │  │
│  │  ┌─────────────┬─────────────┬─────────────┬─────────────────┐ │  │
│  │  │ type       │ length      │ max_length  │ *elements       │ │  │
│  │  │ (4 bytes)  │ (4 bytes)   │ (4 bytes)   │ (8 bytes)       │ │  │
│  │  └─────────────┴─────────────┴─────────────┴─────────────────┘ │  │
│  │                                                                   │  │
│  │  ┌───────────────────────────────────────────────────────────┐ │  │
│  │  │ initial_elements[ ] (内联存储)                             │ │  │
│  │  └───────────────────────────────────────────────────────────┘ │  │
│  └───────────────────────────────────────────────────────────────────┘  │
│                                                                         │
│  当元素较少时:                                                          │
│  elements 指向 initial_elements                                          │
│  ┌─────┬─────┬─────┬─────┬─────┐                                     │
│  │ [0] │ [1] │ [2] │ [3] │ ... │  ListCell 数组                      │
│  └─────┴─────┴─────┴─────┴─────┘                                     │
│                                                                         │
│  当元素较多时:                                                          │
│  elements 指向单独分配的内存                                              │
│  ┌─────┬─────┬─────┬─────┬─────┬─────┬─────┐                         │
│  │ [0] │ [1] │ [2] │ [3] │ [4] │ ... │ [N] │  ListCell 数组         │
│  └─────┴─────┴─────┴─────┴─────┴─────┴─────┘                         │
│                                                                         │
└─────────────────────────────────────────────────────────────────────────┘

2.4 空列表表示

c
// 空列表的唯一合法表示是 NULL
#define NIL ((List *) NULL)

// 判断列表是否为空的正确方式
if (list == NIL)           // 正确
if (list == NULL)          // 也正确
if (list_length(list) == 0) // 正确
if (IsPointerList(list) && list == NIL)  // 显式检查

3. 四种列表类型

类型NodeTag元素类型典型用途
T_List通用指针列表void *节点树、语句元素、表达式列表
T_IntList整数列表int列号、偏移量、索引
T_OidListOID 列表Oid表 OID、类型 OID、函数 OID
T_XidList事务 ID 列表TransactionId事务列表

3.1 类型检查宏

c
#define IsPointerList(l)    ((l) == NIL || IsA((l), List))
#define IsIntegerList(l)    ((l) == NIL || IsA((l), IntList))
#define IsOidList(l)        ((l) == NIL || IsA((l), OidList))
#define IsXidList(l)        ((l) == NIL || IsA((l), XidList))

重要: NIL 被认为是任意类型的空列表!


4. 元素访问宏

4.1 基础访问宏

c
// 从 ListCell 获取值
#define lfirst(lc)             ((lc)->ptr_value)           // 指针值
#define lfirst_int(lc)         ((lc)->int_value)            // 整数值
#define lfirst_oid(lc)         ((lc)->oid_value)            // OID 值
#define lfirst_xid(lc)         ((lc)->xid_value)            // 事务 ID
#define lfirst_node(type, lc)   castNode(type, lfirst(lc))   // 类型化节点

// 从 List 获取值 (按索引)
#define linitial(l)             lfirst(list_nth_cell(l, 0))   // 第 0 个
#define lsecond(l)              lfirst(list_nth_cell(l, 1))   // 第 1 个
#define lthird(l)               lfirst(list_nth_cell(l, 2))   // 第 2 个
#define lfourth(l)              lfirst(list_nth_cell(l, 3))  // 第 3 个
#define llast(l)                lfirst(list_last_cell(l))    // 最后一个

// 按索引获取 (带类型变体)
#define linitial_int(l)        lfirst_int(list_nth_cell(l, 0))
#define linitial_oid(l)        lfirst_oid(list_nth_cell(l, 0))
#define lsecond_int(l)         lfirst_int(list_nth_cell(l, 1))
#define llast_int(l)           lfirst_int(list_last_cell(l))
#define llast_oid(l)           lfirst_oid(list_last_cell(l))

// 类型化节点获取
#define linitial_node(type, l)  castNode(type, linitial(l))
#define lsecond_node(type, l)   castNode(type, lsecond(l))
#define llast_node(type, l)     castNode(type, llast(l))

4.2 内联辅助函数

c
// 获取第 n 个元素 (索引从 0 开始)
static inline ListCell *list_nth_cell(const List *list, int n);

// 获取第 n 个元素的指针值
static inline void *list_nth(const List *list, int n);

// 获取第 n 个元素的整数值
static inline int list_nth_int(const List *list, int n);

// 获取第 n 个元素的 OID 值
static inline Oid list_nth_oid(const List *list, int n);

// 获取第 n 个元素的类型化节点
#define list_nth_node(type, list, n) castNode(type, list_nth(list, n))

// 获取列表长度
static inline int list_length(const List *l);  // nil 返回 0

// 获取头/尾单元格
static inline ListCell *list_head(const List *l);   // 空列表返回 NULL
static inline ListCell *list_tail(const List *l);   // 空列表返回 NULL

// 获取下一个单元格
static inline ListCell *lnext(const List *l, const ListCell *c);

4.3 访问示例

c
List *nodeList;
List *intList;
List *oidList;

// 节点列表访问
Node *first = linitial(nodeList);           // 第 1 个节点
Node *last = llast(nodeList);                // 最后一个节点
Query *q = lsecond_node(Query, nodeList);   // 第 2 个 Query 节点

// 整数列表访问
int firstInt = linitial_int(intList);
int count = list_length(intList);
int third = list_nth_int(intList, 2);

// OID 列表访问
Oid firstOid = linitial_oid(oidList);
Oid lastOid = llast_oid(oidList);

// 单元格遍历
ListCell *cell;
for (cell = list_head(nodeList); cell != NULL; cell = lnext(nodeList, cell))
{
    Node *node = lfirst(cell);
    // 处理 node
}

5. 列表创建

5.1 便捷创建宏

c
// 指针列表创建 (T_List)
#define list_make1(x1)          list_make1_impl(T_List, list_make_ptr_cell(x1))
#define list_make2(x1, x2)      list_make2_impl(...)
#define list_make3(x1, x2, x3)  list_make3_impl(...)
#define list_make4(x1, x2, x3, x4)  list_make4_impl(...)
#define list_make5(x1, x2, x3, x4, x5)  list_make5_impl(...)

// 整数列表创建 (T_IntList)
#define list_make1_int(x1)       list_make1_impl(T_IntList, list_make_int_cell(x1))
#define list_make2_int(x1, x2)  list_make2_impl(...)
// ... 同理 list_make3_int 到 list_make5_int

// OID 列表创建 (T_OidList)
#define list_make1_oid(x1)      list_make1_impl(T_OidList, list_make_oid_cell(x1))
#define list_make2_oid(x1, x2) list_make2_impl(...)
// ... 同理 list_make3_oid 到 list_make5_oid

// 事务 ID 列表创建 (T_XidList)
#define list_make1_xid(x1)      list_make1_impl(T_XidList, list_make_xid_cell(x1))
// ... 同理 list_make2_xid 到 list_make5_xid

5.2 列表创建函数实现

c
// src/backend/nodes/list.c:235-243
List *
list_make1_impl(NodeTag t, ListCell datum1)
{
    List *list = new_list(t, 1);
    list->elements[0] = datum1;
    check_list_invariants(list);
    return list;
}

List *
list_make2_impl(NodeTag t, ListCell datum1, ListCell datum2)
{
    List *list = new_list(t, 2);
    list->elements[0] = datum1;
    list->elements[1] = datum2;
    check_list_invariants(list);
    return list;
}
// ... list_make3_impl 到 list_make5_impl 类似

5.3 列表创建示例

c
// 创建指针列表
List *columns = list_make2(
    makeVar(1, 1, INT4OID, -1, 0, 0),  // id 列
    makeVar(1, 2, TEXTOID, -1, 0, 0)   // name 列
);

// 创建整数列表 (列号)
List *colNumbers = list_make3_int(1, 2, 3);

// 创建 OID 列表 (表 OID)
List *tableOids = list_make2_oid(RelationGetOid(rel1), 
                                 RelationGetOid(rel2));

// 创建空列表后追加
List *dynamicList = NIL;
dynamicList = lappend(dynamicList, makeNode(Query));
dynamicList = lappend(dynamicList, makeNode(InsertStmt));

5.4 内存分配策略

c
/*
 * new_list() 分配策略:
 * 1. 最小分配 8 个 ListCell 单位
 * 2. 按 2 的幂扩展 (8, 16, 32, 64, ...)
 * 3. 短列表的元素与 List 头在同一块内存
 * 4. 长列表的元素单独分配
 */
static List *
new_list(NodeTag type, int min_size)
{
    int max_size = pg_nextpower2_32(Max(8, min_size + LIST_HEADER_OVERHEAD));
    max_size -= LIST_HEADER_OVERHEAD;
    
    newlist = palloc(offsetof(List, initial_elements) + 
                     max_size * sizeof(ListCell));
    // ...
}

6. 列表遍历

6.1 foreach 基础遍历

c
// src/include/nodes/pg_list.h:373-379
#define foreach(cell, lst)  \
    for (ForEachState cell##__state = {(lst), 0};         \
         (cell##__state.l != NIL &&                       \
          cell##__state.i < cell##__state.l->length) ?    \
         (cell = &cell##__state.l->elements[cell##__state.i], true) : \
         (cell = NULL, false);                            \
         cell##__state.i++)

使用示例:

c
List *tableList;
ListCell *cell;

// 基础遍历
foreach (cell, tableList)
{
    RangeVar *rv = (RangeVar *) lfirst(cell);
    printf("Table: %s\n", rv->relname);
}

// 获取当前索引
ListCell *cell;
int index = 0;
foreach (cell, tableList)
{
    RangeVar *rv = (RangeVar *) lfirst(cell);
    printf("[%d] Table: %s\n", index++, rv->relname);
}

// break 后 cell 指向当前元素
foreach (cell, tableList)
{
    RangeVar *rv = (RangeVar *) lfirst(cell);
    if (rv->if_not_exists)
        break;
}
// 此时 cell 仍然有效

6.2 便捷遍历宏

c
// 指针遍历 (推荐写法)
Node *node;
foreach_ptr(Node, node, myList)
{
    // 处理 node
}

// 整数遍历
int num;
foreach_int(num, intList)
{
    printf("Number: %d\n", num);
}

// OID 遍历
Oid oid;
foreach_oid(oid, oidList)
{
    printf("OID: %u\n", oid);
}

// 事务 ID 遍历
TransactionId xid;
foreach_xid(xid, xidList)
{
    printf("XID: %u\n", xid);
}

6.3 foreach_node 类型安全遍历

c
// 自动类型检查的遍历
Query *query;
foreach_node(Query, query, queryList)
{
    // query 已经是正确的类型
    // 如果元素不是 Query 类型,会触发断言失败
}

// 例如遍历目标列表
TargetEntry *tle;
foreach_node(TargetEntry, tle, targetlist)
{
    printf("Result %d: %s\n", tle->resno, tle->resname);
}

6.4 高级遍历

c
// 从指定位置开始遍历
ListCell *cell;
for_each_from(cell, myList, 5)  // 从第 5 个元素开始
{
    Node *node = lfirst(cell);
    // ...
}

// 从指定单元格开始遍历
ListCell *startCell = list_head(myList)->next;  // 从第 2 个开始
for_each_cell(cell, myList, startCell)
{
    Node *node = lfirst(cell);
    // ...
}

// 同时遍历两个列表 (以较短列表为准)
ListCell *cell1, *cell2;
forboth(cell1, list1, cell2, list2)
{
    Node *n1 = lfirst(cell1);
    Node *n2 = lfirst(cell2);
    // ...
}

// 遍历三个/四个/五个列表
ListCell *c1, *c2, *c3;
forthree(c1, list1, c2, list2, c3, list3) { /* ... */ }

ListCell *c1, *c2, *c3, *c4;
forfour(c1, list1, c2, list2, c3, list3, c4, list4) { /* ... */ }

6.5 遍历中的删除操作

c
// 使用 foreach_delete_current 安全删除当前元素
List *filtered = NIL;
ListCell *cell;
foreach (cell, originalList)
{
    Node *node = lfirst(cell);
    if (should_keep(node))
        filtered = lappend(filtered, node);
    else
        filtered = foreach_delete_current(filtered, cell);
}

// 获取当前遍历索引
foreach (cell, myList)
{
    int idx = foreach_current_index(cell);
    printf("Element %d\n", idx);
}

7. 列表操作函数

7.1 追加操作 (Append)

c
// 追加指针到列表末尾
List *lappend(List *list, void *datum);

// 追加整数到列表末尾
List *lappend_int(List *list, int datum);

// 追加 OID 到列表末尾
List *lappend_oid(List *list, Oid datum);

// 追加事务 ID 到列表末尾
List *lappend_xid(List *list, TransactionId datum);

实现原理:

c
// src/backend/nodes/list.c:338-351
List *
lappend(List *list, void *datum)
{
    Assert(IsPointerList(list));
    
    if (list == NIL)
        list = new_list(T_List, 1);   // 空列表创建新列表
    else
        new_tail_cell(list);          // 现有列表追加单元格
    
    llast(list) = datum;             // 设置最后一个元素
    check_list_invariants(list);
    return list;
}

示例:

c
List *list = NIL;

// 追加节点
list = lappend(list, makeNode(Query));
list = lappend(list, makeNode(InsertStmt));
list = lappend(list, makeNode(UpdateStmt));
// list 现在有 3 个元素

// 追加整数
List *nums = NIL;
nums = lappend_int(nums, 10);
nums = lappend_int(nums, 20);
nums = lappend_int(nums, 30);

7.2 前置操作 (Prepend)

c
// 在列表头部添加元素 (O(n) 操作,需要移动所有元素!)
List *lcons(void *datum, List *list);
List *lcons_int(int datum, List *list);
List *lcons_oid(Oid datum, List *list);

示例:

c
List *list = list_make2_int(3, 4);
// [3, 4]

list = lcons_int(2, list);
// [2, 3, 4]

list = lcons_int(1, list);
// [1, 2, 3, 4]

7.3 列表连接 (Concat)

c
// 破坏性连接: 修改 list1,追加 list2 的所有元素
List *list_concat(List *list1, const List *list2);

// 非破坏性连接: 创建新列表
List *list_concat_copy(const List *list1, const List *list2);

示例:

c
List *list1 = list_make2_int(1, 2);
List *list2 = list_make3_int(3, 4, 5);

// 破坏性连接
List *result1 = list_concat(list1, list2);
// result1 = [1, 2, 3, 4, 5]
// list1 被修改,现在也是 [1, 2, 3, 4, 5]

// 非破坏性连接
List *result2 = list_concat_copy(list1, list2);
// result2 = [1, 2, 3, 4, 5]
// list1 保持不变

7.4 列表删除

c
// 按值删除 (使用 equal() 比较)
List *list_delete(List *list, void *datum);
List *list_delete_int(List *list, int datum);
List *list_delete_oid(List *list, Oid datum);

// 按指针删除 (使用 == 比较)
List *list_delete_ptr(List *list, void *datum);

// 删除第 n 个元素
List *list_delete_nth_cell(List *list, int n);

// 按单元格删除
List *list_delete_cell(List *list, ListCell *cell);

// 删除第一个元素
List *list_delete_first(List *list);

// 删除最后一个元素
List *list_delete_last(List *list);

// 删除前 n 个元素
List *list_delete_first_n(List *list, int n);

// 截断列表
List *list_truncate(List *list, int new_size);

示例:

c
List *list = list_make4_int(1, 2, 3, 4);
// [1, 2, 3, 4]

// 删除第 2 个元素 (索引从 0 开始)
list = list_delete_nth_cell(list, 1);
// [1, 3, 4]

// 按值删除
list = list_delete_int(list, 3);
// [1, 4]

// 删除第一个
list = list_delete_first(list);
// [4]

// 截断 (保留前 n 个)
List *nums = list_make5_int(1, 2, 3, 4, 5);
nums = list_truncate(nums, 3);
// [1, 2, 3]

7.5 列表成员检查

c
// 使用 equal() 比较 (结构相等)
bool list_member(const List *list, const void *datum);

// 使用指针比较 (地址相等)
bool list_member_ptr(const List *list, const void *datum);

// 整数列表成员检查
bool list_member_int(const List *list, int datum);

// OID 列表成员检查
bool list_member_oid(const List *list, Oid datum);

// 事务 ID 列表成员检查
bool list_member_xid(const List *list, TransactionId datum);

示例:

c
List *nums = list_make3_int(1, 2, 3);

if (list_member_int(nums, 2))
    printf("Found 2!\n");

if (!list_member_int(nums, 99))
    printf("99 not found!\n");

// 节点列表成员检查 (使用结构相等)
List *queries;
Query *target = makeNode(Query);
if (list_member(queries, target))
    printf("Query found!\n");

7.6 列表操作 (并集/交集/差集)

c
// 并集 (返回新列表)
List *list_union(const List *list1, const List *list2);
List *list_union_ptr(const List *list1, const List *list2);
List *list_union_int(const List *list1, const List *list2);
List *list_union_oid(const List *list1, const List *list2);

// 交集
List *list_intersection(const List *list1, const List *list2);
List *list_intersection_int(const List *list1, const List *list2);

// 差集 (list1 - list2)
List *list_difference(const List *list1, const List *list2);
List *list_difference_ptr(const List *list1, const List *list2);
List *list_difference_int(const List *list1, const List *list2);
List *list_difference_oid(const List *list1, const List *list2);

// 追加唯一元素 (只在不存在时追加)
List *list_append_unique(List *list, void *datum);
List *list_append_unique_ptr(List *list, void *datum);
List *list_append_unique_int(List *list, int datum);
List *list_append_unique_oid(List *list, Oid datum);

// 连接并去重
List *list_concat_unique(List *list1, const List *list2);
List *list_concat_unique_ptr(List *list1, const List *list2);
List *list_concat_unique_int(List *list1, const List *list2);
List *list_concat_unique_oid(List *list1, const List *list2);

示例:

c
List *a = list_make3_int(1, 2, 3);
List *b = list_make3_int(2, 3, 4);

// 并集
List *union = list_union_int(a, b);
// [1, 2, 3, 4]

// 交集
List *inter = list_intersection_int(a, b);
// [2, 3]

// 差集
List *diff = list_difference_int(a, b);
// [1]

// 追加唯一
List *unique = list_make2_int(1, 2);
unique = list_append_unique_int(unique, 2);  // 忽略
unique = list_append_unique_int(unique, 3);  // 添加
// [1, 2, 3]

7.7 列表复制

c
// 浅复制 (复制 List 结构,元素指针不变)
List *list_copy(const List *oldlist);

// 复制前 n 个元素
List *list_copy_head(const List *oldlist, int n);

// 跳过前 n 个元素复制
List *list_copy_tail(const List *oldlist, int n);

// 深复制 (复制 List 结构和所有元素)
List *list_copy_deep(const List *oldlist);

// 复制前示例:
List *original = list_make2_int(1, 2);
List *copy = list_copy(original);
// original 和 copy 长度相同,但指向相同的元素

// 深复制示例:
List *original = list_make1(makeNode(Query));
List *deep = list_copy_deep(original);
// original 和 deep 都指向不同的 Query 节点

7.8 列表排序

c
// 通用排序
void list_sort(List *list, list_sort_comparator cmp);

// 整数排序比较器
int list_int_cmp(const ListCell *p1, const ListCell *p2);

// OID 排序比较器
int list_oid_cmp(const ListCell *p1, const ListCell *p2);

示例:

c
// 排序整数列表
List *nums = list_make5_int(5, 3, 1, 4, 2);
list_sort(nums, list_int_cmp);
// nums 现在是 [1, 2, 3, 4, 5]

// 自定义比较器
int my_cmp(const ListCell *a, const ListCell *b)
{
    return lfirst_int(b) - lfirst_int(a);  // 降序
}
list_sort(nums, my_cmp);
// nums 现在是 [5, 4, 3, 2, 1]

// 排序后去重
list_sort(oidList, list_oid_cmp);
list_deduplicate_oid(oidList);

7.9 列表释放

c
// 释放列表结构 (不释放元素)
void list_free(List *list);

// 释放列表结构和所有元素 (元素必须是 palloc 分配)
void list_free_deep(List *list);

示例:

c
// 释放列表结构
List *temp = list_make3_int(1, 2, 3);
list_free(temp);  // 仅释放 List 结构
// temp 应该设置为 NIL

// 深释放 (当元素是 palloc 分配时)
List *strList = NIL;
strList = lappend(strList, pstrdup("hello"));
strList = lappend(strList, pstrdup("world"));
list_free_deep(strList);  // 释放 List 结构和字符串

// 清理后
temp = NIL;
strList = NIL;

8. 高级使用模式

8.1 构建动态列表

c
// 场景: 根据条件动态构建列表
List *build_column_list(ColumnInfo *cols, int ncols)
{
    List *result = NIL;
    
    for (int i = 0; i < ncols; i++)
    {
        if (cols[i].include)
        {
            TargetEntry *tle = makeTargetEntry(
                makeVar(1, cols[i].attno, cols[i].type, -1, 0, 0),
                i + 1,
                cols[i].name,
                false
            );
            result = lappend(result, tle);
        }
    }
    
    return result;
}

8.2 过滤列表

c
// 过滤出特定类型的节点
List *
filter_expressions(List *all_nodes)
{
    List *exprs = NIL;
    ListCell *cell;
    
    foreach (cell, all_nodes)
    {
        Node *node = lfirst(cell);
        if (IsA(node, Var) || IsA(node, Const) || IsA(node, FuncExpr))
        {
            exprs = lappend(exprs, node);
        }
    }
    
    return exprs;
}

8.3 列表反转

c
// 反转列表顺序
List *
reverse_list(List *list)
{
    List *reversed = NIL;
    ListCell *cell;
    
    foreach (cell, list)
    {
        reversed = lcons(lfirst(cell), reversed);
    }
    
    return reversed;
}

8.4 列表映射

c
// 对列表中每个元素应用转换函数
List *
map_list(List *input, Node *(*transform)(Node *))
{
    List *result = NIL;
    ListCell *cell;
    
    foreach (cell, input)
    {
        Node *original = lfirst(cell);
        Node *transformed = transform(original);
        result = lappend(result, transformed);
    }
    
    return result;
}

// 使用示例
Node *double_const(Node *node)
{
    if (IsA(node, Const))
    {
        Const *c = (Const *) node;
        if (c->consttype == INT4OID && !c->constisnull)
        {
            Const *newc = makeNode(Const);
            *newc = *c;
            newc->constvalue = Int32GetDatum(DatumGetInt32(c->constvalue) * 2);
            return (Node *) newc;
        }
    }
    return node;
}

8.5 列表累加

c
// 累加计算
int
sum_list(List *nums)
{
    int sum = 0;
    int num;
    
    foreach_int(num, nums)
    {
        sum += num;
    }
    
    return sum;
}

// 查找满足条件的元素
Node *
find_node(List *nodes, bool (*pred)(Node *))
{
    ListCell *cell;
    
    foreach (cell, nodes)
    {
        Node *node = lfirst(cell);
        if (pred(node))
            return node;
    }
    
    return NULL;
}

// 使用示例
bool is_select(Node *node)
{
    return IsA(node, SelectStmt);
}

Query *sel = (Query *) find_node(stmtList, is_select);

8.6 多列表同步遍历

c
// 场景: 同时遍历两个相关列表
void process_columns(List *names, List *types, List *defaults)
{
    ListCell *nameCell, *typeCell, *defaultCell;
    
    forboth(nameCell, names, typeCell, types)
    {
        char *name = strVal(lfirst(nameCell));
        Oid type = lfirst_oid(typeCell);
        
        // 处理默认值 (可能为空列表)
        if (defaults != NIL)
        {
            Node *def = lfirst(defaults);
            // ...
        }
    }
}

9. 性能注意事项

9.1 操作复杂度

操作复杂度说明
lappendO(1) 均摊追加到末尾
lconsO(n)前置需要移动所有元素
list_concatO(n)连接两个列表
list_delete_nthO(n)删除需要移动后续元素
list_memberO(n)线性查找
list_lengthO(1)直接返回长度字段

9.2 最佳实践

c
// ✓ 推荐: 使用 lappend 构建列表
List *items = NIL;
for (i = 0; i < n; i++)
    items = lappend(items, create_item(i));

// ✗ 避免: 频繁在列表头部插入
List *items = NIL;
for (i = 0; i < n; i++)
    items = lcons(create_item(i), items);  // O(n²)!

// ✓ 推荐: 反转后使用 lappend
List *items = reverse_list(build_in_order());
// 或者使用栈式构建再反转

// ✓ 推荐: 如果需要频繁在两端操作,考虑其他数据结构
// List 不是双向队列的最佳选择

// ✓ 推荐: 避免长列表的线性搜索
// 对于大列表,考虑使用哈希表或数组

9.3 内存使用

c
/*
 * 内存分配特点:
 * 1. 小列表 (< 8 元素) 在 List 头附近分配
 * 2. 元素使用 2 的幂扩展
 * 3. 删除操作不会立即释放多余的内存
 * 4. list_truncate 不会释放多余内存
 */

// 释放未使用的内存
List *compact_list(List *list)
{
    return list_copy(list);  // 只复制需要的元素
}

10. 调试技巧

10.1 打印列表内容

c
// 使用 nodeToString 打印节点列表
#include "nodes/pg_list.h"
#include "utils/elog.h"

void print_list(List *list, const char *name)
{
    elog(DEBUG1, "%s has %d elements", name, list_length(list));
    
    ListCell *cell;
    int i = 0;
    foreach (cell, list)
    {
        Node *node = lfirst(cell);
        char *str = nodeToString(node);
        elog(DEBUG1, "  [%d]: %s", i++, str);
        pfree(str);
    }
}

10.2 断言检查

c
// 开发时启用列表不变量检查
#ifdef USE_ASSERT_CHECKING
static void check_list_invariants(const List *list);
#endif

// 在操作后检查
void my_list_operation(List *list)
{
    // ... 操作 ...
    check_list_invariants(list);  // 仅在断言模式下有效
}

10.3 Valgrind 支持

c
// 编译时启用 DEBUG_LIST_MEMORY_USAGE
// 可以检测列表操作后的无效指针访问
#ifdef USE_VALGRIND
#define DEBUG_LIST_MEMORY_USAGE
#endif

11. 源码文件索引

文件路径内容描述
src/include/nodes/pg_list.hList 头文件: 结构、宏、内联函数
src/backend/nodes/list.cList 实现: 所有操作函数
src/include/nodes/nodes.hNode 基结构
src/include/nodes/nodetags.hNodeTag 枚举 (含 T_List 等)

11.1 关键函数速查

函数用途
list_make1/2/3/4/5创建固定长度列表
lappend/lappend_int/lappend_oid追加元素
lcons/lcons_int/lcons_oid前置元素
list_concat/list_concat_copy连接列表
list_delete/list_delete_nth_cell删除元素
list_member/list_member_int/list_member_oid成员检查
list_copy/list_copy_deep复制列表
list_sort/list_int_cmp/list_oid_cmp排序
list_free/list_free_deep释放列表
foreach/foreach_ptr/foreach_node遍历宏
forboth/forthree/forfour多列表遍历

12. 总结

12.1 核心要点

  1. 可扩展数组: List 使用连续内存数组而非链表,提供更好的缓存局部性

  2. 四种类型: T_List(指针)、T_IntList(整数)、T_OidList(OID)、T_XidList(事务 ID)

  3. 空列表为 NIL: 唯一合法的空列表表示是 NULL (即 NIL)

  4. 操作复杂度: lappend O(1),lcons O(n),list_delete O(n)

  5. 类型安全: 使用专用函数处理不同类型列表,避免类型混淆

  6. 内存优化: 小列表使用内联存储,避免小内存分配

12.2 命名约定

约定示例说明
l 前缀lappend, lconsList 操作
_int 后缀lappend_int, list_member_int整数列表
_oid 后缀lappend_oid, list_member_oidOID 列表
_xid 后缀lappend_xid事务 ID 列表
lfist_ 前缀lfirst, lfirst_int从单元格获取
linitial/lsecond/...快捷访问从列表获取指定位置

12.3 与传统链表的对比

特性PostgreSQL List传统链表
内存布局连续数组分散节点
缓存友好性
前置操作O(n)O(1)
追加操作O(1) 均摊O(1)
删除第 n 个O(n)O(n)
随机访问O(1)O(n)
内存开销高 (每个节点需要指针)

本文档基于 PostgreSQL 源码 (v17) 分析编写,详细阐述了 PostgreSQL List 数据结构的设计、实现和使用方法。