Postgresql实验系列（3）最简脏读插件

• 本文参考社区插件实现一个最简脏读函数，实现脏读功能（读取已经删除或未提交数据）。
• 脏读的核心逻辑只有一个关键词：SnapshotAny。

效果

drop table foo;
CREATE TABLE foo (bar bigint, baz text);
ALTER TABLE foo SET (autovacuum_enabled = false, toast.autovacuum_enabled = false);
INSERT INTO foo VALUES (1, 'Hello world');
SELECT * FROM dirtyread('foo') as t(bar bigint, baz text);
DELETE FROM foo;
SELECT * FROM dirtyread('foo') as t(bar bigint, baz text);
1
2
3
4
5
6
7

SRF关键逻辑&数据结构分析

1. SRF_IS_FIRSTCALL

插件系统为返回多行的插件提供了SRF框架，函数会被自动反复调用多次，直到SRF_RETURN_DONE。

2. SRF_FIRSTCALL_INIT

为fn_extra填充数据FuncCallContext，红色部分：

3. MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);

切换到SRF专用内存上下文。

4. get_call_result_type

拿到输出元组格式，例如当前用例：

SELECT * FROM dirtyread('foo') as t(bar bigint, baz text);
1

会得到tupdesc：

{natts = 2, tdtypeid = 2249, tdtypmod = 0, tdrefcount = -1, constr = 0x0, attrs = 0x1523858}

attrs[0] = {attrelid = 0, attname = {data = "bar", ...
attrs[1] = {attrelid = 0, attname = {data = "baz", ...
1
2
3
4

5. funcctx->tuple_desc = BlessTupleDesc(tupdesc)

记录输出元组格式

(gdb) p *(FuncCallContext*)fcinfo->flinfo->fn_extra
$32 = {call_cntr = 0, max_calls = 0, user_fctx = 0x0, attinmeta = 0x0, multi_call_memory_ctx = 0x1524c00, tuple_desc = 0x1523840}
1
2

修改蓝色部分：

6. funcctx->user_fctx = (void *)inter_call_data;

记录用户自定义数据结构，修改紫色部分：

7. SRF_PERCALL_SETUP

拿到上图中FuncCallContext。

8. SRF_RETURN_NEXT

修改计数器与状态，粉色部分：

code

#include "postgres.h"

#include "access/heapam.h"
#include "access/htup_details.h"
#include "access/table.h"
#include "access/transam.h"
#include "access/xact.h"
#include "access/xlog.h"
#include "access/xlog.h"
#include "catalog/pg_type.h"
#include "common/hashfn.h"
#include "fmgr.h"
#include "funcapi.h"
#include "miscadmin.h"
#include "storage/proc.h"
#include "storage/procarray.h"
#include "utils/builtins.h"
#include "utils/datum.h"
#include "utils/lsyscache.h"
#include "utils/rel.h"
#include "utils/snapmgr.h"
#include "utils/syscache.h"

Datum dirtyread(PG_FUNCTION_ARGS);

typedef struct dirtyread_ctx_state
{
	Relation		rel;
	TupleDesc		desc;
	TableScanDesc	scan;
} dirtyread_ctx_state;

PG_MODULE_MAGIC;

PG_FUNCTION_INFO_V1(dirtyread);
Datum dirtyread(PG_FUNCTION_ARGS)
{
	FuncCallContext *funcctx;
	dirtyread_ctx_state *inter_call_data = NULL;
	HeapTuple tuple;

	if (SRF_IS_FIRSTCALL())
	{
		MemoryContext oldcontext;
		Oid relid;
		TupleDesc tupdesc;

		relid = PG_GETARG_OID(0);
		funcctx = SRF_FIRSTCALL_INIT();
		oldcontext = MemoryContextSwitchTo(funcctx->multi_call_memory_ctx);

		inter_call_data = (dirtyread_ctx_state *)palloc(sizeof(dirtyread_ctx_state));
		inter_call_data->rel = table_open(relid, AccessShareLock);
		inter_call_data->desc = RelationGetDescr(inter_call_data->rel);

		if (get_call_result_type(fcinfo, NULL, &tupdesc) != TYPEFUNC_COMPOSITE)
			elog(ERROR, "return type must be a row type");
		funcctx->tuple_desc = BlessTupleDesc(tupdesc);

		inter_call_data->scan = heap_beginscan(inter_call_data->rel, SnapshotAny, 0, NULL, NULL, 0);

		funcctx->user_fctx = (void *)inter_call_data;

		MemoryContextSwitchTo(oldcontext);
	}

	funcctx = SRF_PERCALL_SETUP();
	inter_call_data = (dirtyread_ctx_state *)funcctx->user_fctx;

	if ((tuple = heap_getnext(inter_call_data->scan, ForwardScanDirection)) != NULL)
	{
		SRF_RETURN_NEXT(funcctx, heap_copy_tuple_as_datum(tuple, inter_call_data->desc));
	}
	else
	{
		heap_endscan(inter_call_data->scan);
		table_close(inter_call_data->rel, AccessShareLock);
		SRF_RETURN_DONE(funcctx);
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80