庖丁解牛——从 BRIN Index 到 PostgreSQL AM

typedef struct BrinMetaPageData
{
	uint32      brinMagic;      /* Magic Number */
	uint32      brinVersion;    /* 版本号 */
	BlockNumber pagesPerRange;  /* 建表时指定的 pages per range */
	BlockNumber lastRevmapPage; /* 最后一页 Range Map */
} BrinMetaPageData;

Datum
brinhandler(PG_FUNCTION_ARGS)
{
	IndexAmRoutine *amroutine = makeNode(IndexAmRoutine);

	/*......*/

	amroutine->ambuild = brinbuild;                 /* 创建索引时该怎么做 */
	amroutine->ambuildempty = brinbuildempty;       /* 创建空索引时该怎么做 */
	amroutine->aminsert = brininsert;               /* 当heap有tuple插入时该怎么做 */
	amroutine->ambulkdelete = brinbulkdelete;       /* 批量删除tuple时该怎么做 */
	amroutine->amvacuumcleanup = brinvacuumcleanup; /* vacuum的时候该怎么做 */
	amroutine->amcostestimate = brincostestimate;   /* 成本估计 */
	amroutine->amoptions = brinoptions;             /* 解析pages_per_range和autosummarize参数 */
	amroutine->amvalidate = brinvalidate;           /* 校验函数 */
	amroutine->ambeginscan = brinbeginscan;         /* begin scan */
	amroutine->amrescan = brinrescan;               /* re scan */
	amroutine->amgettuple = NULL;                   /* BRIN是对Page的总结，所以没有点查 */
	amroutine->amgetbitmap = bringetbitmap;         /* bitmap index scan */
	amroutine->amendscan = brinendscan;             /* end scan */

	PG_RETURN_POINTER(amroutine);
}

int64 bringetbitmap(IndexScanDesc scan, Node **bmNodeP) {

	/* 打开 heap table */
	heapOid = IndexGetRelation(RelationGetRelid(idxRel), false);
	heapRel = table_open(heapOid, AccessShareLock);
	/* 获取 heap file 有多少个 page */
	nblocks = RelationGetNumberOfBlocks(heapRel);

	/* 从第 0 个 page 开始读取，每隔 pages per range 读取一个 page
	 * 这里不从 rang map 读取的原因就在于 range map 可能会发生并发更新，从而导致读取出来的数据不对 */
	for (heapBlk = 0; heapBlk < nblocks; heapBlk += opaque->bo_pagesPerRange) {
		bool		addrange;
		bool		gottuple = false;
		BrinTuple  *tup;
		OffsetNumber off;
		Size		size;

		/* 获取 Brin Tuple，也就是从 Revmap 中读取到 Brin Tuple 的位置，然后再把 Brin Tuple 读出来 */
		tup = brinGetTupleForHeapBlock(opaque->bo_rmAccess, heapBlk, &buf,
				&off, &size, BUFFER_LOCK_SHARE, scan->xs_snapshot);
		if (tup) {
			gottuple = true;
			btup = brin_copy_tuple(tup, size, btup, &btupsz);
			LockBuffer(buf, BUFFER_LOCK_UNLOCK);
		}

		if (!gottuple)
			addrange = true;
		else {
			/* 获取 BRIN Tuple，将其转换成 BrinMemTuple */
			dtup = brin_deform_tuple(bdesc, btup, dtup);
			if (dtup->bt_placeholder)
				addrange = true;
			else {
				int keyno;
				/* 根据扫描 key 进行对比 */
				addrange = true;
				for (keyno = 0; keyno < scan->numberOfKeys; keyno++) {
					/* ...... */
				}
			}
		}

		/* 如果当前 BRIN Tuple 符合筛选条件的话，那么就需要将其所对应的
		 * pages 添加至结果中，也就是 bmNodeP 中 */
		if (addrange) {
			BlockNumber pageno;
			/* 遍历所有的 pages */
			for (pageno = heapBlk; pageno <= heapBlk + opaque->bo_pagesPerRange - 1; pageno++) {
				MemoryContextSwitchTo(oldcxt);
				tbm_add_page(tbm, pageno);
				totalpages++;
				MemoryContextSwitchTo(perRangeCxt);
			}
		}
	}
}

bool
brininsert(Relation idxRel, Datum *values, bool *nulls,
		   ItemPointer heaptid, Relation heapRel,
		   IndexUniqueCheck checkUnique,
		   IndexInfo *indexInfo)
{
	revmap = brinRevmapInitialize(idxRel, &pagesPerRange, NULL);

	/* origHeapBlk 表示发生 tuple 插入的 heap page.  heapBlk
	 * heapBlk 则表示该 heap page 所在组的第一个 heap page */
	origHeapBlk = ItemPointerGetBlockNumber(heaptid);
	heapBlk = (origHeapBlk / pagesPerRange) * pagesPerRange;

	for (;;) {
		/* 在建立索引时指定了 autosummarize 为 true，同时插入的 tuple page 是
		 * summarize 的第一个 page */
		if (autosummarize && heapBlk > 0 && heapBlk == origHeapBlk &&
			ItemPointerGetOffsetNumber(heaptid) == FirstOffsetNumber) {
			BlockNumber lastPageRange = heapBlk - 1;
			BrinTuple  *lastPageTuple;
			/* 获取上一页的 Brin Tuple */
			lastPageTuple = brinGetTupleForHeapBlock(revmap, lastPageRange, &buf, &off,
						NULL, BUFFER_LOCK_SHARE, NULL);
			if (!lastPageTuple) {
			    /* 发送 vacuum 请，类型为 AVW_BRINSummarizeRange */
				recorded = AutoVacuumRequestWork(AVW_BRINSummarizeRange,
								RelationGetRelid(idxRel), lastPageRange);
			}
			else
				LockBuffer(buf, BUFFER_LOCK_UNLOCK);
		}

		/* 获取 BRIN Tuple */
		brtup = brinGetTupleForHeapBlock(revmap, heapBlk, &buf, &off, NULL, BUFFER_LOCK_SHARE, NULL);

		/* if range is unsummarized, there's nothing to do */
		/*
		 * 注意这个地方，如果当前 range 没有进行 summarize 的话，将会直接跳过。
		 * 那么为什么不在这里进行 summarize 呢? 原因在于 summarize 是一个很耗时的过程，需要扫描 pages per range 个
		 * page 从而找到 max/min value，这里 PG 将其设计成异步更新，即 summarize 的过程只由 vacuum 进程来做。
		 * 这样一来索引数据的插入不会影响 heap data 的插入效率 */
		if (!brtup)
			break;

		dtup = brin_deform_tuple(bdesc, brtup, NULL);

		/* 逐一对 Index Key 进行比较，如果该值大于最大值或者是小于最小值的话，需要更新 BRIN Tuple values */
		for (keyno = 0; keyno < bdesc->bd_tupdesc->natts; keyno++)
		{
			/* ...... */
			/* if that returned true, we need to insert the updated tuple */
			need_insert |= DatumGetBool(result);
		}

		if (!need_insert) 
			LockBuffer(buf, BUFFER_LOCK_UNLOCK);
		else {
			/* 更新 BRIN Tuple */
			if (!brin_doupdate(idxRel, pagesPerRange, revmap, heapBlk,
							   buf, off, origtup, origsz, newtup, newsz,
							   samepage, false)) {
				/* no luck; start over */
				MemoryContextResetAndDeleteChildren(tupcxt);
				continue;
			}
		}
		/* success! */
		break;
	}
	return false;
}