PostgreSQL中PageAddItemExtended函数的逻辑是什么

本篇内容主要讲解“postgresql中PageAddItemExtended函数的逻辑是什么”，感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷，实用性强。下面就让小编来带大家学习“Postgresql中PageAddItemExtended函数的逻辑是什么”吧!

一、测试数据表

testdb=# drop table if exists t_insert;
NOTICE:  table "t_insert" does not exist, skipping
DROP TABLE
testdb=# create table t_insert(id int,c1 char(10),c2 char(10),c3 char(10));
CREATE TABLE

二、源码分析

插入数据主要的实现在bufpage.c中，主要的函数是PageAddItemExtended。该函数的英文注释已经说得非常清楚，不过为了更好理解，这里添加了中文注释。
变量、宏定义和结构体等说明

1、Page
指向char的指针：
typedef char *Pointer;
typedef Pointer Page;
2、Item
指向char的指针：
typedef Pointer Item;
3、Size
typedef size_t  Size;
4、OffsetNumber
unsigned short（16bits）：
typedef uint16 OffsetNumber;
5、PageHeader
PageHeaderData结构体指针
typedef struct PageHeaderData
 {
     
     PageXLogRecPtr pd_lsn;      
     uint16      pd_checksum;    
     uint16      pd_flags;       
     LocationIndex pd_lower;     
     LocationIndex pd_upper;     
     LocationIndex pd_special;   
     uint16      pd_pagesize_version;
     TransactionId pd_prune_xid; 
     ItemIdData  pd_linp[FLEXIBLE_ARRAY_MEMBER]; 
 } PageHeaderData;
 
 typedef PageHeaderData *PageHeader;

#define PD_HAS_FREE_LINES   0x0001  
#define PD_PAGE_FULL        0x0002  
#define PD_ALL_VISIBLE      0x0004  
#define PD_VALID_FLAG_BITS  0x0007  

6、SizeOfPageHeaderData
长整型，定义页头大小：
#define offsetof(type, field)   ((long) &((type *)0)->field)
 #define SizeOfPageHeaderData (offsetof(PageHeaderData, pd_linp))
7、BLCKSZ
#define BLCKSZ 8192
8、PageGetMaxOffsetNumber
如果空闲空间的lower小于等于页头大小，值为0，否则值为lower减去页头大小（24Bytes）后除以ItemId大小（4Bytes）
 #define PageGetMaxOffsetNumber(page) \
     (((PageHeader) (page))->pd_lower <= SizeOfPageHeaderData ? 0 : \
      ((((PageHeader) (page))->pd_lower - SizeOfPageHeaderData) \
       / sizeof(ItemIdData)))
9、InvalidOffsetNumber
是否无效的偏移值
 #define InvalidOffsetNumber     ((OffsetNumber) 0)
10、OffsetNumberNext
输入值+1
 #define OffsetNumberNext(offsetNumber) \
     ((OffsetNumber) (1 + (offsetNumber)))
11、ItemId
结构体ItemIdData指针
 typedef struct ItemIdData
 {
     unsigned    lp_off:15,      
                 lp_flags:2,     
                 lp_len:15;      
 } ItemIdData;
 
 typedef ItemIdData *ItemId;

 #define LP_UNUSED 0 
 #define LP_NORMAL 1 
 #define LP_REDIRECT 2 
 #define LP_DEAD 3 

12、PageGetItemId
获取相应的ItemIdData的指针
#define PageGetItemId(page, offsetNumber) \
    ((ItemId) (&((PageHeader) (page))->pd_linp[(offsetNumber) - 1]))
13、Pai_OVERWRITE
位标记，实际值为1
#define PAI_OVERWRITE           (1 << 0)
14、PAI_IS_HEAP
位标记，实际值为2
#define PAI_IS_HEAP             (1 << 1)
15、ItemIdIsUsed
ItemId是否已被占用
 #define ItemIdIsUsed(itemId) \
     ((itemId)->lp_flags != LP_UNUSED)
16、ItemIdHasStorage
 #define ItemIdHasStorage(itemId) \
     ((itemId)->lp_len != 0
17、PageHasFreeLinePointers/PageClearHasFreeLinePointers
判断Page从Header到Lower之间是否有空位
#define PageHasFreeLinePointers(page) \
    (((PageHeader) (page))->pd_flags & PD_HAS_FREE_LINES)
清除空位标记（标记错误时清除）
#define PageClearHasFreeLinePointers(page) \
    (((PageHeader) (page))->pd_flags &= ~PD_HAS_FREE_LINES)
18、MaxHeapTuplesPerPage
每个Page可以容纳的最大Tuple数，计算公式是：
（Block大小 - 页头大小） / (对齐后行头部大小 + 行指针大小）
#define MaxHeapTuplesPerPage    \
     ((int) ((BLCKSZ - SizeOfPageHeaderData) / \
             (MAXALIGN(SizeofHeapTupleHeader) + sizeof(ItemIdData))))

PageAddItemExtended函数解读

OffsetNumber
PageAddItemExtended(Page page,
                    Item item,
                    Size size,
                    OffsetNumber offsetNumber,
                    int flags)
{
    PageHeader  phdr = (PageHeader) page;//页头指针
    Size        alignedSize;//对齐大小
    int         lower;//Free space低位
    int         upper;//Free space高位
    ItemId      itemId;//行指针
    OffsetNumber limit;//行偏移，Free space中第1个可用的位置偏移
    bool        needshuffle = false;//是否需要移动原有数据

    
    if (phdr->pd_lower < SizeOfPageHeaderData ||
        phdr->pd_lower > phdr->pd_upper ||
        phdr->pd_upper > phdr->pd_special ||
        phdr->pd_special > BLCKSZ)
        ereport(PANIC,
                (errcode(ERRCODE_DATA_CORRUPTED),
                 errmsg("corrupted page pointers: lower = %u, upper = %u, special = %u",
                        phdr->pd_lower, phdr->pd_upper, phdr->pd_special)));

    
        //获取存储数据的偏移（位于Lower和Upper之间）
    limit = OffsetNumberNext(PageGetMaxOffsetNumber(page));

    
    if (OffsetNumberIsValid(offsetNumber))
    {
        //如果指定了数据存储的偏移量（传入的偏移量参数有效）
        
        if ((flags & PAI_OVERWRITE) != 0) //不覆盖原有数据
        {
            if (offsetNumber < limit)
            {
                //获取指定偏移的ItemId
                itemId = PageGetItemId(phdr, offsetNumber);
                //指定的数据偏移已使用或者已分配存储空间，报错
                if (ItemIdIsUsed(itemId) || ItemIdHasStorage(itemId))
                {
                    elog(WARNING, "will not overwrite a used ItemId");
                    return InvalidOffsetNumber;
                }
            }
        }
        else//覆盖原有数据
        {
            //指定的行偏移不在空闲空间中，需要移动原数据为新数据腾位置
            if (offsetNumber < limit)
                needshuffle = true; 
        }
    }
    else//没有指定数据存储行偏移
    {
        
        
        if (PageHasFreeLinePointers(phdr))//页头标记提示存在已回收的空间
        {
            
            //循环找出第1个可用的空闲行偏移
            for (offsetNumber = 1; offsetNumber < limit; offsetNumber++)
            {
                itemId = PageGetItemId(phdr, offsetNumber);
                if (!ItemIdIsUsed(itemId) && !ItemIdHasStorage(itemId))
                    break;
            }
            //没有找到，说明页头标记有误，需清除标记，以免误导
            if (offsetNumber >= limit)
            {
                
                PageClearHasFreeLinePointers(phdr);
            }
        }
        else//没有已回收的空间，行指针/数据存储到Free Space中
        {
            
            offsetNumber = limit;
        }
    }

    
    if (offsetNumber > limit)
    {
        //如果指定的偏移大于空闲空间可用的第1个位置，报错
        elog(WARNING, "specified item offset is too large");
        return InvalidOffsetNumber;
    }

    
    if ((flags & PAI_IS_HEAP) != 0 && offsetNumber > MaxHeapTuplesPerPage)
    {
        //Heap数据，但偏移大于一页可存储的最大Tuple数，报错
        elog(WARNING, "can't put more than MaxHeapTuplesPerPage items in a heap page");
        return InvalidOffsetNumber;
    }

    
    if (offsetNumber == limit || needshuffle)
        //如果数据存储在Free space中，修改lower值
        lower = phdr->pd_lower + sizeof(ItemIdData);
    else
        lower = phdr->pd_lower;//否则，找到了已回收的空闲位置，使用原有的lower

    alignedSize = MAXALIGN(size);//大小对齐

    upper = (int) phdr->pd_upper - (int) alignedSize;//申请存储空间

    if (lower > upper)//校验
        return InvalidOffsetNumber;

    
    //获取行指针
    itemId = PageGetItemId(phdr, offsetNumber);
    //
    if (needshuffle)
        //如果需要腾位置，把原有的行指针往后挪一"格"
        memmove(itemId + 1, itemId,
                (limit - offsetNumber) * sizeof(ItemIdData));

    
    //设置新数据行指针
    ItemIdSetNormal(itemId, upper, size);

    
    VALGRIND_CHECK_MEM_IS_DEFINED(item, size);

    
    //把数据放在数据区
    memcpy((char *) page + upper, item, size);

    
    //更新页头的lower & upper
    phdr->pd_lower = (LocationIndex) lower;
    phdr->pd_upper = (LocationIndex) upper;
    //如成功，返回实际的行偏移
    return offsetNumber;
}

三、跟踪分析

下面使用gdb对PageAddItemExtended函数跟踪分析。
测试场景：首先插入8行数据，然后删除第2行，接着插入1行数据：

testdb=# -- 插入8行数据
testdb=# insert into t_insert values(1,'11','12','13');
insert into t_insert values(6,'11','12','13');
insert into t_insert values(7,'11','12','13');
insert into t_insert values(8,'11','12','13');

checkpoint;INSERT 0 1
testdb=# insert into t_insert values(2,'11','12','13');
INSERT 0 1
testdb=# insert into t_insert values(3,'11','12','13');
INSERT 0 1
testdb=# insert into t_insert values(4,'11','12','13');
INSERT 0 1
testdb=# insert into t_insert values(5,'11','12','13');
INSERT 0 1
testdb=# insert into t_insert values(6,'11','12','13');
INSERT 0 1
testdb=# insert into t_insert values(7,'11','12','13');
INSERT 0 1
testdb=# insert into t_insert values(8,'11','12','13');
INSERT 0 1
testdb=# 
testdb=# checkpoint;
CHECKPOINT
testdb=# -- 删除第2行
testdb=# delete from t_insert where id = 2;
DELETE 1
testdb=# 
testdb=# checkpoint;
CHECKPOINT
testdb=# 
testdb=# -- 获取pid
testdb=# select pg_backend_pid();
 pg_backend_pid 
----------------
           1572
(1 row)

使用gdb跟踪最后插入1行数据的过程：

#启动gdb，绑定进程
[root@localhost demo]# gdb -p 1572
GNU gdb (GDB) Red Hat Enterprise linux 7.6.1-100.el7
Copyright (C) 2013 Free Software Foundation, Inc.
License GPLv3+: GNU GPL version 3 or later <Http://gnu.org/licenses/gpl.html>
This is free software: you are free to change and Redistribute it.
There is NO WARRANTY, to the extent permitted by law.  Type "show copying"
and "show warranty" for details.
This GDB was configured as "x86_64-redhat-linux-gnu".
For bug reporting instructions, please see:
<http://www.gnu.org/software/gdb/bugs/>.
Attaching to process 1572
...
(gdb) 
#设置断点
(gdb) b PageAddItemExtended
Breakpoint 1 at 0x845119: file bufpage.c, line 196.
(gdb) 

#切换到psql，执行插入语句
testdb=# insert into t_insert values(9,'11','12','13');
（挂起）

#切换回gdb
(gdb) c
Continuing.

Breakpoint 1, PageAddItemExtended (page=0x7feaaefac300 "\001", item=0x29859f8 "2\234\030", size=61, offsetNumber=0, flags=2) at bufpage.c:196
196     PageHeader  phdr = (PageHeader) page;
#断点在函数的第一行代码上
#bt打印调用栈
(gdb) bt
#0  PageAddItemExtended (page=0x7feaaefac300 "\001", item=0x29859f8 "2\234\030", size=61, offsetNumber=0, flags=2) at bufpage.c:196
#1  0x00000000004cf4f9 in RelationPutHeapTuple (relation=0x7feac6e2ccb8, buffer=141, tuple=0x29859e0, token=false) at hio.c:53
#2  0x00000000004c34ec in heap_insert (relation=0x7feac6e2ccb8, tup=0x29859e0, cid=0, options=0, bistate=0x0) at heapam.c:2487
#3  0x00000000006c076b in ExecInsert (mtstate=0x2984c10, slot=0x2985250, planSlot=0x2985250, estate=0x29848c0, canSetTag=true) at nodeModifyTable.c:529
#4  0x00000000006c29f3 in ExecModifyTable (pstate=0x2984c10) at nodeModifyTable.c:2126
#5  0x000000000069a7d8 in ExecProcNodeFirst (node=0x2984c10) at execProcnode.c:445
#6  0x0000000000690994 in ExecProcNode (node=0x2984c10) at ../../../src/include/executor/executor.h:237
#7  0x0000000000692e5e in ExecutePlan (estate=0x29848c0, planstate=0x2984c10, use_parallel_mode=false, operation=CMD_INSERT, sendTuples=false, numberTuples=0, direction=ForwardScanDirection, 
    dest=0x2990dc8, execute_once=true) at execMain.c:1726
#8  0x0000000000690e58 in standard_ExecutorRun (queryDesc=0x2981020, direction=ForwardScanDirection, count=0, execute_once=true) at execMain.c:363
#9  0x0000000000690cef in ExecutorRun (queryDesc=0x2981020, direction=ForwardScanDirection, count=0, execute_once=true) at execMain.c:306
#10 0x0000000000851d84 in ProcessQuery (plan=0x2990c68, sourceText=0x28c5ef0 "insert into t_insert values(9,'11','12','13');", params=0x0, queryEnv=0x0, dest=0x2990dc8, completionTag=0x7ffdbc052d10 "")
    at pquery.c:161
#11 0x00000000008534f4 in PortalRunMulti (portal=0x292b490, isTopLevel=true, setHoldSnapshot=false, dest=0x2990dc8, altdest=0x2990dc8, completionTag=0x7ffdbc052d10 "") at pquery.c:1286
#12 0x0000000000852b32 in PortalRun (portal=0x292b490, count=9223372036854775807, isTopLevel=true, run_once=true, dest=0x2990dc8, altdest=0x2990dc8, completionTag=0x7ffdbc052d10 "") at pquery.c:799
#13 0x000000000084cebc in exec_simple_query (query_string=0x28c5ef0 "insert into t_insert values(9,'11','12','13');") at postgres.c:1122
#14 0x0000000000850f3c in PostgresMain (arGC=1, argv=0x28efaa8, dbname=0x28ef990 "testdb", username=0x28ef978 "xdb") at postgres.c:4153
#15 0x00000000007c0168 in BackendRun (port=0x28e7970) at postmaster.c:4361
#16 0x00000000007bf8fc in BackendStartup (port=0x28e7970) at postmaster.c:4033
#17 0x00000000007bc139 in ServerLoop () at postmaster.c:1706
#18 0x00000000007bb9f9 in PostmasterMain (argc=1, argv=0x28c0b60) at postmaster.c:1379
#19 0x00000000006f19e8 in main (argc=1, argv=0x28c0b60) at main.c:228

#使用next命令，单步调试
(gdb) next
207     if (phdr->pd_lower < SizeOfPageHeaderData ||
(gdb) next
208         phdr->pd_lower > phdr->pd_upper ||
(gdb) next
207     if (phdr->pd_lower < SizeOfPageHeaderData ||
(gdb) next
209         phdr->pd_upper > phdr->pd_special ||
(gdb) next
208         phdr->pd_lower > phdr->pd_upper ||
(gdb) next
210         phdr->pd_special > BLCKSZ)
(gdb) next
209         phdr->pd_upper > phdr->pd_special ||
(gdb) next
219     limit = OffsetNumberNext(PageGetMaxOffsetNumber(page));
(gdb) next
222     if (OffsetNumberIsValid(offsetNumber))
(gdb) p offsetNumber
$1 = 0
#offsetNumber为0，没有指定插入的行偏移
(gdb) next
247         if (PageHasFreeLinePointers(phdr))
#查看页头信息
(gdb) p *phdr
$2 = {pd_lsn = {xlogid = 1, xrecoff = 3677462648}, pd_checksum = 0, pd_flags = 0, pd_lower = 56, pd_upper = 7680, pd_special = 8192, pd_pagesize_version = 8196, pd_prune_xid = 1612849, 
  pd_linp = 0x7feaaefac318}
(gdb) 
#查看行指针信息
(gdb) p phdr->pd_linp[0]
$3 = {lp_off = 8128, lp_flags = 1, lp_len = 61}
(gdb) p phdr->pd_linp[1]
$4 = {lp_off = 8064, lp_flags = 1, lp_len = 61}
...
(gdb) p phdr->pd_linp[8]
$11 = {lp_off = 0, lp_flags = 0, lp_len = 0}
#行指针偏移的lp_flags均为1（LP_NORMAL），表示正在使用。
...
298     alignedSize = MAXALIGN(size);
(gdb) p size
$17 = 61
(gdb) p alignedSize
$18 = 5045956
(gdb) next
300     upper = (int) phdr->pd_upper - (int) alignedSize;
(gdb) p alignedSize
$19 = 64 #原为61，对齐为64
(gdb) next
302     if (lower > upper)
(gdb) p lower
$20 = 60
(gdb) p upper
$21 = 7616
(gdb) next
308     itemId = PageGetItemId(phdr, offsetNumber);
(gdb) 
310     if (needshuffle)
(gdb) next
315     ItemIdSetNormal(itemId, upper, size);
(gdb) next
332     memcpy((char *) page + upper, item, size);
(gdb) p *itemId
$23 = {lp_off = 7616, lp_flags = 1, lp_len = 61}
...
(gdb) next
338     return offsetNumber;
(gdb) 
#数据插入在行偏移为9（行指针下标=8）的位置上。
(gdb) p offsetNumber
$24 = 9
(gdb) p phdr->pd_linp[8]
$25 = {lp_off = 7616, lp_flags = 1, lp_len = 61}
...
#说明：之所以新插入的数据没有放在2号偏移上面，是因为被删除的第2行没有被回收。
#查看Page中的数据
testdb=# select * from heap_page_items(get_raw_page('t_insert',0));
 lp | lp_off | lp_flags | lp_len | t_xmin  | t_xmax  | t_field3 | t_ctid | t_infomask2 | t_infomask | t_hoff | t_bits | t_oid |                                    t_data                                 
   
----+--------+----------+--------+---------+---------+----------+--------+-------------+------------+--------+--------+-------+---------------------------------------------------------------------------
---
  1 |   8128 |        1 |     61 | 1612841 |       0 |        0 | (0,1)  |           4 |       2306 |     24 |        |       | \x010000001731312020202020202020173132202020202020202017313320202020202020
20
  2 |   8064 |        1 |     61 | 1612842 | 1612849 |        0 | (0,2)  |        8196 |        258 |     24 |        |       | \x020000001731312020202020202020173132202020202020202017313320202020202020
20
  3 |   8000 |        1 |     61 | 1612843 |       0 |        0 | (0,3)  |           4 |       2306 |     24 |        |       | \x030000001731312020202020202020173132202020202020202017313320202020202020
20
  4 |   7936 |        1 |     61 | 1612844 |       0 |        0 | (0,4)  |           4 |       2306 |     24 |        |       | \x040000001731312020202020202020173132202020202020202017313320202020202020
20
  5 |   7872 |        1 |     61 | 1612845 |       0 |        0 | (0,5)  |           4 |       2306 |     24 |        |       | \x050000001731312020202020202020173132202020202020202017313320202020202020
20
  6 |   7808 |        1 |     61 | 1612846 |       0 |        0 | (0,6)  |           4 |       2306 |     24 |        |       | \x060000001731312020202020202020173132202020202020202017313320202020202020
20
  7 |   7744 |        1 |     61 | 1612847 |       0 |        0 | (0,7)  |           4 |       2306 |     24 |        |       | \x070000001731312020202020202020173132202020202020202017313320202020202020
20
  8 |   7680 |        1 |     61 | 1612848 |       0 |        0 | (0,8)  |           4 |       2306 |     24 |        |       | \x080000001731312020202020202020173132202020202020202017313320202020202020
20
  9 |   7616 |        1 |     61 | 1612850 |       0 |        0 | (0,9)  |           4 |       2050 |     24 |        |       | \x090000001731312020202020202020173132202020202020202017313320202020202020
20
(9 rows)

到此，相信大家对“PostgreSQL中PageAddItemExtended函数的逻辑是什么”有了更深的了解，不妨来实际操作一番吧！这里是编程网网站，更多相关内容可以进入相关频道进行查询，关注我们，继续学习！