技术实践丨PostgreSQL插件之pg_dirtyread "闪回查询"

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

 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 ); --测试方便，先把自动vacuum关闭掉。 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

 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

SELECT * FROM pg_dirtyread('foo') AS t(tableoid oid, ctid tid, xmin xid, xmax xid, cmin cid, cmax cid, dead boolean, bar bigint, baz text); tableoid │ ctid │ xmin │ xmax │ cmin │ cmax │ dead │ bar │ baz ──────────┼───────┼──────┼──────┼──────┼──────┼──────┼─────┼─────────────────── 41823 │ (0,1) │ 1484 │ 1485 │ 0 │ 0 │ t │ 1 │ Delete 41823 │ (0,2) │ 1484 │ 0 │ 0 │ 0 │ f │ 2 │ Insert 41823 │ (0,3) │ 1484 │ 1486 │ 0 │ 0 │ t │ 3 │ Update 41823 │ (0,4) │ 1484 │ 1488 │ 0 │ 0 │ f │ 4 │ Not deleted 41823 │ (0,5) │ 1484 │ 1489 │ 1 │ 1 │ f │ 5 │ Not updated 41823 │ (0,6) │ 1486 │ 0 │ 0 │ 0 │ f │ 3 │ Updated 41823 │ (0,7) │ 1489 │ 0 │ 1 │ 1 │ t │ 5 │ Not quite updated 41823 │ (0,8) │ 1490 │ 0 │ 2 │ 2 │ t │ 6 │ Not inserted

HeapTuple dirtyread_do_convert_tuple(HeapTuple tuple, TupleConversionMap *map, TransactionId oldest_xmin) { /* * Extract all the values of the old tuple, offsetting the arrays so that * invalues[0] is left NULL and invalues[1] is the first source attribute; * this exactly matches the numbering convention in attrMap. */ heap_deform_tuple(tuple, map->indesc, invalues + 1, inisnull + 1); //+1是因为是从下标1开始，从旧的元组中把数据的值获取到 /* * Transpose into proper fields of the new tuple. 这部分是重点，在这里完成转换 */ for (i = 0; i < outnatts; i++) { int j = attrMap; if (j == DeadFakeAttributeNumber) //场景1：明确是dead，直接调用内核的函数HeapTupleIsSurelyDead即可， //定义在tqual.c中，其它场景可以使用HeapTupleSatisfiesVacuum、HeapTupleSatisfiesMVCC等等，这里明确是dead，所以使用HeapTupleIsSurelyDead { outvalues = HeapTupleIsSurelyDead(tuple , oldest_xmin); outisnull = false; } else if (j < 0) //场景2：系统列，交给函数heap_getsysattr来处理。 outvalues = heap_getsysattr(tuple, j, map->indesc, &outisnull); else { //场景3：最常见的场景，直接获取即可。 outvalues = invalues[j]; outisnull = inisnull[j]; } } return heap_form_tuple(map->outdesc, outvalues, outisnull); //重新包装为tuple格式 }

 if (SRF_IS_FIRSTCALL())，这部分比较套路化 { superuser校验 PG_GETARG_OID获取表的oid heap_open打开表 get_call_result_type计算结果校验，不支持复合类型 BlessTupleDesc(tupdesc) 拿到表结构 usr_ctx->map = dirtyread_convert_tuples_by_name(usr_ctx->reltupdesc, funcctx->tuple_desc, "Error converting tuple descriptors!"); //关键的一步，这里使用dirtyread_convert_tuples_by_name函数，。 heap_beginscan(usr_ctx->rel, SnapshotAny...),开始启动表扫描，这里使用了SnapshotAny }

 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); //结束扫描 heap_close(usr_ctx->rel, AccessShareLock); //关闭表 SRF_RETURN_DONE(funcctx); }