PgSQL - 内核插件 - pg_dirtyread

yzsDBA

发布于 2024-04-12 16:05:37

1380

发布于 2024-04-12 16:05:37

文章被收录于专栏：PostgreSQL研究与原理解析PostgreSQL研究与原理解析

PgSQL - 内核插件 - pg_dirtyread

表中删除了记录，并且没有进行vacuum，此时可以通过pg_dirtyread扩展读取死记录。

1、使用方法

CREATE EXTENSION pg_dirtyread;
SELECT * FROM pg_dirtyread('tablename') AS t(col1 type1, col2 type2, ...);

安装插件后，通过pg_dirtyread函数读取所有记录，包括已删除且没有被vacuum的记录。函数的入参为表名，因为该函数返回RECORD，所以需要使用AS指定表的别名，同时指定读取的列及其列类型。注：pg_dirtyread入参使用表的OID也可以，当然若使用表名则会在代码中转换成表的OID。

举例：

CREATE EXTENSION pg_dirtyread;
-- Create table and disable autovacuum
CREATE TABLE foo (bar bigint, baz text);
ALTER TABLE foo SET (
autovacuum_enabled = false, toast.autovacuum_enabled = false
);
INSERT INTO foo VALUES (1, 'Test'), (2, 'New Test');
DELETE FROM foo WHERE bar = 1;
SELECT * FROM pg_dirtyread('foo') as t(bar bigint, baz text);
bar │   baz
────┼──────────
  1 │ Test
  2 │ New Test

也可以读取删除列的内容，当前前提是没有执行VACUUM FULL或CLUSTER重写表。使用dropped_N来访问第N列，值从1开始。PG删除了原始列的类型信息，因此如果在表别名中指定了正确的类型，则仅能进行一些健全性检测：类型长度、类型对其方式、类型修饰符和传递值：

CREATE TABLE ab(a text, b text);
INSERT INTO ab VALUES ('Hello', 'World');
ALTER TABLE ab DROP COLUMN b;
DELETE FROM ab;
SELECT * FROM pg_dirtyread('ab') ab(a text, dropped_2 text);
  a   │ dropped_2
──────┼───────────
Hello │ World

系统列比如max和ctid也可以通过在表别名中指定进行检索。有一个特殊的列dead可以报告该行值是否是死记录（HeapTupleIsSurelyDead函数判断），当然这个列不能在恢复中使用，也就是备机使用不了。oid列在PG11及其之后版本使用：

2、原理

pg_dirtyread.c主要是面向用户使用的API函数接口pg_dirtyread的实现：

Datum
pg_dirtyread(PG_FUNCTION_ARGS)
{
  if (SRF_IS_FIRSTCALL()){
    {//会话第一次调用会初始化一些信息
        if (!superuser())//只能是超级用户使用
            ereport(ERROR,
                    (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
                     errmsg("must be superuser to use pg_dirtyread")));
    //通过表OID得到表的tuple描述符
    relid = PG_GETARG_OID(0);//函数第一个入参即为表OID，若是表名则会转换成表OID
        funcctx = SRF_FIRSTCALL_INIT();
        oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);
        usr_ctx = (pg_dirtyread_ctx *) palloc(sizeof(pg_dirtyread_ctx));
        usr_ctx->rel =table_open(relid, AccessShareLock);//打开表
        usr_ctx->reltupdesc = RelationGetDescr(usr_ctx->rel);//获取表的tuple描述符TupleDesc
    //不支持复合类型，表别名定义的结构得到输出记录的tupdesc
        if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
            ereport(ERROR,
                    (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
                     errmsg("function returning record called in context "
                         "that cannot accept type record")));
    //得到表别名完整的tuple描述符
        funcctx->tuple_desc = BlessTupleDesc(tupdesc);
    //关键的一步，这里使用dirtyread_convert_tuples_by_name：得到表别名和表定义的字段映射
        usr_ctx->map = dirtyread_convert_tuples_by_name(usr_ctx->reltupdesc,
                funcctx->tuple_desc, "Error converting tuple descriptors!");
    //开始启动扫描表，所有记录都可见
        usr_ctx->scan = heap_beginscan(usr_ctx->rel, SnapshotAny,...);//使用SnapshotAny
        /* only call GetOldestXmin while not in recovery */
        if (!RecoveryInProgress())
            usr_ctx->oldest_xmin = GetOldestXmin(usr_ctx->rel , 0);
        funcctx->user_fctx = (void *) usr_ctx;
        MemoryContextSwitchTo(oldcontext);
    }
    funcctx = SRF_PERCALL_SETUP();
    usr_ctx = (pg_dirtyread_ctx *) funcctx->user_fctx;
  //不断获取每一行，然后对每一行进行转换，直到扫描结束
    if ((tuplein = heap_getnext(usr_ctx->scan, ForwardScanDirection)) != NULL)
    {
        if (usr_ctx->map != NULL)
        {//根据映射进行记录字段调整，输出记录给用户
            tuplein = dirtyread_do_convert_tuple(tuplein, usr_ctx->map, usr_ctx->oldest_xmin);
            SRF_RETURN_NEXT(funcctx, HeapTupleGetDatum(tuplein));
        }
        else
            SRF_RETURN_NEXT(funcctx, heap_copy_tuple_as_datum(tuplein, usr_ctx->reltupdesc));
    }
    else
    {
        heap_endscan(usr_ctx->scan);
        table_close(usr_ctx->rel, AccessShareLock);
        SRF_RETURN_DONE(funcctx);
    }
}

可见性判断函数：