PostgreSQL源码学习插入数据#4，5

Z时代
2024-01-10
分类：综合

database

本节介绍table_tuple_insert和ExecInsert函数。

table_tuple_insert函数

//src/include/access/tableam.h
static inline void
table_tuple_insert(Relation rel, TupleTableSlot *slot, CommandId cid,
				   int options, struct BulkInsertStateData *bistate)
{
	/* 默认的表存储引擎的话，rd_tableam赋值为heapam_methods，
	   其tuple_insert指向的就是上一节讲过的heapam_tuple_insert函数 */
	rel->rd_tableam->tuple_insert(rel, slot, cid, options,
							  bistate);
}

ExecInsert函数

//src/backend/executor/nodeModifyTable.c
/* 物化slot */
ExecMaterializeSlot(slot);
/* 取得result relation相关信息 */
resultRelInfo = estate->es_result_relation_info;
resultRelationDesc = resultRelInfo->ri_RelationDesc;
/* 执行BEFORE ROW INSERT触发器 */
if (resultRelInfo->ri_TrigDesc &&
	resultRelInfo->ri_TrigDesc->trig_insert_before_row)
{
	if (!ExecBRInsertTriggers(estate, resultRelInfo, slot))
		return NULL;		/* "do nothing" */
}
/* 如果设置了INSTEAD OF ROW触发器，执行触发器 */
if (resultRelInfo->ri_TrigDesc &&
	resultRelInfo->ri_TrigDesc->trig_insert_instead_row)
{
	if (!ExecIRInsertTriggers(estate, resultRelInfo, slot))
		return NULL;		/* "do nothing" */
}
/* 若没有INSTEAD OF ROW触发器，但是是外部表 */
else if (resultRelInfo->ri_FdwRoutine)
{
	/* 如果有生成列的话先计算生成列 */
	if (resultRelationDesc->rd_att->constr &&
		resultRelationDesc->rd_att->constr->has_generated_stored)
		ExecComputeStoredGenerated(estate, slot);
	/* 执行外部表的insert流程 */
	slot = resultRelInfo->ri_FdwRoutine->ExecForeignInsert(estate,
								   resultRelInfo,
								   slot,
								   planSlot);
	if (slot == NULL)		/* "do nothing" */
		return NULL;
	/* AFTER ROW触发器或RETURNING约束表达式可能会用到tuple的tableoid，所以这里提前给它赋值一下 */
	slot->tts_tableOid = RelationGetRelid(resultRelInfo->ri_RelationDesc);
}
/* 若没有INSTEAD OF ROW触发器，也不是外部表，走一般流程 */
else
{
	/* 可能有约束会用到tableoid，所以提前对它进行赋值 */
	slot->tts_tableOid = RelationGetRelid(resultRelationDesc);
	/* 如果有的话计算生成列的数据 */
	if (resultRelationDesc->rd_att->constr &&
		resultRelationDesc->rd_att->constr->has_generated_stored)
		ExecComputeStoredGenerated(estate, slot);、
	/* 根据操作类型确定需要执行哪种行安全策略检查 */
		wco_kind = (mtstate->operation == CMD_UPDATE) ?
		WCO_RLS_UPDATE_CHECK : WCO_RLS_INSERT_CHECK;
	/* 执行Policy中的WithCheckOptions检查，在此函数中只进行wco_kind指定的类型检查 */
	if (resultRelInfo->ri_WithCheckOptions != NIL)
		ExecWithCheckOptions(wco_kind, resultRelInfo, slot, estate);
	/* 执行tuple的约束检查（例如NOT NULL这种） */
	if (resultRelationDesc->rd_att->constr)
		ExecConstraints(resultRelInfo, slot, estate);
	/* 若有分区表约束，执行分区表约束检查 */
	if (resultRelInfo->ri_PartitionCheck &&
		(resultRelInfo->ri_PartitionRoot == NULL ||
		 (resultRelInfo->ri_TrigDesc &&
		  resultRelInfo->ri_TrigDesc->trig_insert_before_row)))
		ExecPartitionCheck(resultRelInfo, slot, estate, true);
	/* 语句中含有ON CONFLICT ...， 并且表拥有索引时的情况 */
	if (onconflict != ONCONFLICT_NONE && resultRelInfo->ri_NumIndices > 0)
	{
		/* 获取执行约束检查的索引列表 */
		arbiterIndexes = resultRelInfo->ri_onConflictArbiterIndexes;
vlock:	/* goto返回点 */
		/* 先在这些索引列表上执行一个粗略的冲突检查 */
		specConflict = false;
		if (!ExecCheckIndexConstraints(slot, estate, &conflictTid,
							   arbiterIndexes))
		{
			/* 若检查到冲突，并且语句为ON CONFLICT...DO UPDATE */
			if (onconflict == ONCONFLICT_UPDATE)
			{
				/* 执行ON CONFLICT的UPDATE */
				if (ExecOnConflictUpdate(mtstate, resultRelInfo,
							 &conflictTid, planSlot, slot,
							 estate, canSetTag, &returning))
					{
						/* 计数+1，返回结果 */
						InstrCountTuples2(&mtstate->ps, 1);
						return returning;
					}
				else
					/* 若没有成功执行UPDATE，回到前面重新开始 */
					goto vlock;
			}
			/* 若有冲突，语句是ON CONFLICT DO NOTHING */
			else
			{
				/* 如同语句意思什么也不做，但是需要验证一下在更高的隔离级别中，
				   元组在MVCC快照中对执行器是可见的 */
				Assert(onconflict == ONCONFLICT_NOTHING);
				ExecCheckTIDVisible(estate, resultRelInfo, &conflictTid,
						ExecGetReturningSlot(estate, resultRelInfo));
				/* 计数+1，返回NULL */
				InstrCountTuples2(&mtstate->ps, 1);
				return NULL;
			}
		}
		/* 在开始插入之前，先获取“推测插入锁”，其它流程可以通过它
		   来等待我们决定是否插入，而不用等待我们整个事务结束 */
		specToken = SpeculativeInsertionLockAcquire(GetCurrentTransactionId());
		/* 插入数据，默认存储引擎的话，这里调用heapam_tuple_insert_speculative函数 */
		table_tuple_insert_speculative(resultRelationDesc, slot,
							   estate->es_output_cid,
							   0,
							   NULL,
							   specToken);
		/* 为新插入的元组更新索引 */
		recheckIndexes = ExecInsertIndexTuples(slot, estate, true,
							   &specConflict,
							   arbiterIndexes);
		/* 调整元组的状态，默认存储引擎的话，这里调用heapam_tuple_complete_speculative函数 */
		table_tuple_complete_speculative(resultRelationDesc, slot,
							 specToken, !specConflict);
		/* 唤醒其它在等待此元组决定是否插入的流程，可以根据这里的选择继续流程了 */
		SpeculativeInsertionLockRelease(GetCurrentTransactionId());
		/* 如果有冲突，跳到前面重新开始，因为在ExecInsertIndexTuples流程中
		   已经把有冲突的项加进去了，所以再次检测的话可以发现冲突 */
		if (specConflict)
		{
			list_free(recheckIndexes);
			goto vlock;
		}
		/* 若没有冲突，说明已经成功完成了ON CONFLICT的流程 */
	}
	/* 没有ON CONFLICT语句的话，走正常插入流程 */
	else
	{
		/* 插入元组，上面讲了此函数 */
		table_tuple_insert(resultRelationDesc, slot,
					   estate->es_output_cid,
					   0, NULL);
		if (resultRelInfo->ri_NumIndices > 0)
			recheckIndexes = ExecInsertIndexTuples(slot, estate, false, NULL,
										   NIL);
	}
}
/* 更新计数和last tid */
if (canSetTag)
{
	(estate->es_processed)++;
	setLastTid(&slot->tts_tid);
}
/* 如果本次插入是由于分区键更新带来的，需要执行额外流程 */
ar_insert_trig_tcs = mtstate->mt_transition_capture;
if (mtstate->operation == CMD_UPDATE && mtstate->mt_transition_capture
	&& mtstate->mt_transition_capture->tcs_update_new_table)
{
	ExecARUpdateTriggers(estate, resultRelInfo, NULL,
						 NULL,
						 slot,
						 NULL,
						 mtstate->mt_transition_capture);
	ar_insert_trig_tcs = NULL;
}
/* 执行AFTER ROW INSERT触发器 */
ExecARInsertTriggers(estate, resultRelInfo, slot, recheckIndexes,
				 ar_insert_trig_tcs);
list_free(recheckIndexes);
/* 检查父视图的WITH CHECK OPTIONS约束 */
if (resultRelInfo->ri_WithCheckOptions != NIL)
	ExecWithCheckOptions(WCO_VIEW_CHECK, resultRelInfo, slot, estate);
/* 处理RETURNING */
if (resultRelInfo->ri_projectReturning)
	result = ExecProcessReturning(resultRelInfo, slot, planSlot);
return result;