1 总结
-
portal与事务有强绑定的关系，由portal->createSubid变量记录关联关系。如果为1表示顶层事务，关联的是子事务。
-
PLpgSQL中的提交回滚，有较大限制：
-
子事务ID只增不减，可能有空隙存在，参考PushTransaction的currentSubTransactionId。
2 提交
2.1 顶层事务提交：PreCommit_Portals与子事务提交AtSubCommit_Portals
实例
CREATE OR REPLACE PROCEDURE tproc1()
AS $$
DECLARE
curs1 refcursor;
curs2 CURSOR FOR SELECT c1 FROM tf1;
curs3 CURSOR (key integer) FOR SELECT * FROM tf1 WHERE c1 > key;
x int;
y tf1%ROWTYPE;
BEGIN
open curs1 FOR SELECT * FROM tf1 WHERE c1 > 3;
fetch curs1 into y; RAISE NOTICE 'curs1 : %', y.c3;
fetch curs1 into y; RAISE NOTICE 'curs1 : %', y.c3;
open curs2;
fetch curs2 into x; RAISE NOTICE 'curs2 : %', x;
fetch curs2 into x; RAISE NOTICE 'curs2 : %', x;
OPEN curs3(4); -- OPEN curs3(key := 4);
fetch curs3 into y; RAISE NOTICE 'curs3 : %', y.c4;
fetch curs3 into y; RAISE NOTICE 'curs3 : %', y.c4;
EXCEPTION WHEN others THEN
RAISE NOTICE 'in caller exception';
END;
$$ LANGUAGE plpgsql;
begin;
savepoint sp1;
savepoint sp2;
call tproc1();
commit;
（1）第一次子事务提交：EXCEPTION子事务提交AtSubCommit_Portals（subid=4）
发生在call tproc1();执行完毕。
exec_stmt_block
ReleaseCurrentSubTransaction
CommitSubTransaction
AtSubCommit_Portals
当前事务堆栈：
CurrentTransactionState->subTransactionId
[1, 2, 3, 4]
| | | |
top savepoint savepoint exception
由于call进入SPI，当前SPI堆栈只有一层，对应到3号子事务上，因为是在3号子事务的背景下执行的CALL。
_SPI_stack[0] = {connectSubid = 3}
AtSubCommit_Portals(mySubid=4) 开始处理portal：
portal1 = { // 游标 mySubid = 4 ，当前游标的portal被继承给parent portal
name = "<unnamed portal 17>",
createSubid = 4, // AtSubCommit_Portals 4-->3
activeSubid = 4,
createLevel = 4,
sourceText = "SELECT c1 FROM tf1",
commandTag = CMDTAG_SELECT,
strategy = PORTAL_ONE_SELECT}
… 三个游标的portal都是类似上面处理的。特殊的是顶层portal。不属于当前要释放的子事务4，不处理。
portal1 = { // 顶层 mySubid = 3 ，当前不处理
name = "",
createSubid = 3,
activeSubid = 3,
createLevel = 3,
sourceText = "call tproc1();",
commandTag = CMDTAG_CALL,
strategy = PORTAL_MULTI_QUERY}
（2）第二次子事务提交：savepoint子事务提交AtSubCommit_Portals（subid=3）
发生在commit。
exec_simple_query
finish_xact_command
CommitTransactionCommand
CommitSubTransaction
AtSubCommit_Portals
当前事务堆栈：
CurrentTransactionState->subTransactionId
[1, 2, 3]
| | |
top savepoint savepoint
AtSubCommit_Portals(mySubid=3)开始处理portal：
portal1 = { // 游标 mySubid = 4 ，当前游标的portal被继承给parent portal
name = "<unnamed portal 17>",
createSubid = 2, // AtSubCommit_Portals 3-->2
activeSubid = 2, // AtSubCommit_Portals 3
createLevel = 2, // AtSubCommit_Portals 3-->2
sourceText = "SELECT c1 FROM tf1",
commandTag = CMDTAG_SELECT,
strategy = PORTAL_ONE_SELECT}
当前堆栈Portal已经被drop了。现在PortalHashTable里面只有三个游标的Portal。
（3）第三次子事务提交：savepoint子事务提交（subid=2）
同上。
（4）顶层事务提交（subid=1）PreCommit_Portals
发生在commit。
exec_simple_query
finish_xact_command
CommitTransactionCommand
CommitTransaction
PreCommit_Portals
调用PortalDrop全部释放掉。
PreCommit_Portals函数需要关注的就是，普通portal都会被drop掉。特殊保留的是hold cursor即循环语句使用的内部自建游标，还有一种就是vacuum等多事务语句。
3 回滚
3.1 顶层事务清理：AtAbort_Portals
调用位置
AbortTransaction → AtAbort_Portals
调用一次即可，用户清理顶层事务。
清理逻辑
- 遍历PortalHashTable，拿到所有CreatePortal创建出来的Portal（两种：执行器的portal和游标的portal）
- 标记portal状态failed，标记failed的内存会被释放掉。
- 情况一：elog FATAL 主动标记failed
- 情况二：状态是PORTAL_READY的portal
- 跳过一些portal，不处理。
3.2 子事务清理：AtSubAbort_Portals
调用位置
AbortSubTransaction → AtSubAbort_Portals
有两种调用场景：
清理逻辑
4 PushTransaction与PopTransaction函数
- 启动子事务时需要将当前事务入栈，CurrentTransactionState换成子事务的。
- 子事务和父事务由parent连接。
- 修改这部分代码需要注意一次弹出多个事务时，currentSubTransactionId有没有正确维护。
static void
PushTransaction(void)
{
TransactionState p = CurrentTransactionState;
TransactionState s;
s = (TransactionState)
MemoryContextAllocZero(TopTransactionContext,
sizeof(TransactionStateData));
注意currentSubTransactionId直增不减。pop时也不减少。
currentSubTransactionId += 1;
if (currentSubTransactionId == InvalidSubTransactionId)
{
currentSubTransactionId -= 1;
pfree(s);
ereport(ERROR,
(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
errmsg("cannot have more than 2^32-1 subtransactions in a transaction")));
}
/*
* We can now stack a minimally valid subtransaction without fear of
* failure.
*/
s->fullTransactionId = InvalidFullTransactionId; /* until assigned */
s->subTransactionId = currentSubTransactionId;
s->parent = p;
s->nestingLevel = p->nestingLevel + 1;
s->gucNestLevel = NewGUCNestLevel();
s->savepointLevel = p->savepointLevel;
s->state = TRANS_DEFAULT;
s->blockState = TBLOCK_SUBBEGIN;
GetUserIdAndSecContext(&s->prevUser, &s->prevSecContext);
s->prevXactReadOnly = XactReadOnly;
s->parallelModeLevel = 0;
s->topXidLogged = false;
CurrentTransactionState = s;
}
