gist.c

PostgreSQL 8.1.4的源码 适用于Linux下的开源数据库系统

			ptr->buffer = ReadBuffer(r, ptr->blkno);
			ptr->page = (Page) BufferGetPage(ptr->buffer);
			ptr = ptr->parent;
		}

		/* install new chain of parents to stack */
		child->parent = parent;
		parent->child = child;

		/* make recursive call to normal processing */
		gistFindCorrectParent(r, child);
	}

	return;
}

void
gistmakedeal(GISTInsertState *state, GISTSTATE *giststate)
{
	int			is_splitted;
	ItemId		iid;
	IndexTuple	oldtup,
				newtup;

	/* walk up */
	while (true)
	{
		/*
		 * After this call: 1. if child page was splited, then itup contains
		 * keys for each page 2. if  child page wasn't splited, then itup
		 * contains additional for adjustment of current key
		 */

		if (state->stack->parent)
		{
			/*
			 * X-lock parent page before proceed child, gistFindCorrectParent
			 * should find and lock it
			 */
			gistFindCorrectParent(state->r, state->stack);
		}
		is_splitted = gistplacetopage(state, giststate);

		/* parent locked above, so release child buffer */
		LockBuffer(state->stack->buffer, GIST_UNLOCK);
		ReleaseBuffer(state->stack->buffer);

		/* pop parent page from stack */
		state->stack = state->stack->parent;

		/* stack is void */
		if (!state->stack)
			break;

		/*
		 * child did not split, so we can check is it needed to update parent
		 * tuple
		 */
		if (!is_splitted)
		{
			/* parent's tuple */
			iid = PageGetItemId(state->stack->page, state->stack->childoffnum);
			oldtup = (IndexTuple) PageGetItem(state->stack->page, iid);
			newtup = gistgetadjusted(state->r, oldtup, state->itup[0], giststate);

			if (!newtup)
			{					/* not need to update key */
				LockBuffer(state->stack->buffer, GIST_UNLOCK);
				break;
			}

			state->itup[0] = newtup;
		}
	}							/* while */

	/* release all parent buffers */
	while (state->stack)
	{
		ReleaseBuffer(state->stack->buffer);
		state->stack = state->stack->parent;
	}

	/* say to xlog that insert is completed */
	if (state->needInsertComplete && !state->r->rd_istemp)
		gistxlogInsertCompletion(state->r->rd_node, &(state->key), 1);
}

static void
gistToRealOffset(OffsetNumber *arr, int len, OffsetNumber *reasloffset)
{
	int			i;

	for (i = 0; i < len; i++)
		arr[i] = reasloffset[arr[i]];
}

/*
 *	gistSplit -- split a page in the tree.
 */
IndexTuple *
gistSplit(Relation r,
		  Buffer buffer,
		  IndexTuple *itup,		/* contains compressed entry */
		  int *len,
		  SplitedPageLayout **dist,
		  GISTSTATE *giststate)
{
	Page		p;
	Buffer		leftbuf,
				rightbuf;
	Page		left,
				right;
	IndexTuple *lvectup,
			   *rvectup,
			   *newtup;
	BlockNumber lbknum,
				rbknum;
	GISTPageOpaque opaque;
	GIST_SPLITVEC v;
	GistEntryVector *entryvec;
	int			i,
				fakeoffset,
				nlen;
	OffsetNumber *realoffset;
	IndexTuple *cleaneditup = itup;
	int			lencleaneditup = *len;

	p = (Page) BufferGetPage(buffer);
	opaque = GistPageGetOpaque(p);

	/*
	 * The root of the tree is the first block in the relation.  If we're
	 * about to split the root, we need to do some hocus-pocus to enforce this
	 * guarantee.
	 */
	if (BufferGetBlockNumber(buffer) == GIST_ROOT_BLKNO)
	{
		leftbuf = gistNewBuffer(r);
		GISTInitBuffer(leftbuf, opaque->flags & F_LEAF);
		lbknum = BufferGetBlockNumber(leftbuf);
		left = (Page) BufferGetPage(leftbuf);
	}
	else
	{
		leftbuf = buffer;
		/* IncrBufferRefCount(buffer); */
		lbknum = BufferGetBlockNumber(buffer);
		left = (Page) PageGetTempPage(p, sizeof(GISTPageOpaqueData));
	}

	rightbuf = gistNewBuffer(r);
	GISTInitBuffer(rightbuf, opaque->flags & F_LEAF);
	rbknum = BufferGetBlockNumber(rightbuf);
	right = (Page) BufferGetPage(rightbuf);

	/* generate the item array */
	realoffset = palloc((*len + 1) * sizeof(OffsetNumber));
	entryvec = palloc(GEVHDRSZ + (*len + 1) * sizeof(GISTENTRY));
	entryvec->n = *len + 1;

	fakeoffset = FirstOffsetNumber;
	for (i = 1; i <= *len; i++)
	{
		Datum		datum;
		bool		IsNull;

		if (!GistPageIsLeaf(p) && GistTupleIsInvalid(itup[i - 1]))
		{
			entryvec->n--;
			/* remember position of invalid tuple */
			realoffset[entryvec->n] = i;
			continue;
		}

		datum = index_getattr(itup[i - 1], 1, giststate->tupdesc, &IsNull);
		gistdentryinit(giststate, 0, &(entryvec->vector[fakeoffset]),
					   datum, r, p, i,
					   ATTSIZE(datum, giststate->tupdesc, 1, IsNull),
					   FALSE, IsNull);
		realoffset[fakeoffset] = i;
		fakeoffset++;
	}

	/*
	 * if it was invalid tuple then we need special processing. If it's
	 * possible, we move all invalid tuples on right page. We should remember,
	 * that union with invalid tuples is a invalid tuple.
	 */
	if (entryvec->n != *len + 1)
	{
		lencleaneditup = entryvec->n - 1;
		cleaneditup = (IndexTuple *) palloc(lencleaneditup * sizeof(IndexTuple));
		for (i = 1; i < entryvec->n; i++)
			cleaneditup[i - 1] = itup[realoffset[i] - 1];

		if (gistnospace(left, cleaneditup, lencleaneditup))
		{
			/* no space on left to put all good tuples, so picksplit */
			gistUserPicksplit(r, entryvec, &v, cleaneditup, lencleaneditup, giststate);
			v.spl_leftvalid = true;
			v.spl_rightvalid = false;
			gistToRealOffset(v.spl_left, v.spl_nleft, realoffset);
			gistToRealOffset(v.spl_right, v.spl_nright, realoffset);
		}
		else
		{
			/* we can try to store all valid tuples on one page */
			v.spl_right = (OffsetNumber *) palloc(entryvec->n * sizeof(OffsetNumber));
			v.spl_left = (OffsetNumber *) palloc(entryvec->n * sizeof(OffsetNumber));

			if (lencleaneditup == 0)
			{
				/* all tuples are invalid, so moves half of its to right */
				v.spl_leftvalid = v.spl_rightvalid = false;
				v.spl_nright = 0;
				v.spl_nleft = 0;
				for (i = 1; i <= *len; i++)
					if (i - 1 < *len / 2)
						v.spl_left[v.spl_nleft++] = i;
					else
						v.spl_right[v.spl_nright++] = i;
			}
			else
			{
				/*
				 * we will not call gistUserPicksplit, just put good tuples on
				 * left and invalid on right
				 */
				v.spl_nleft = lencleaneditup;
				v.spl_nright = 0;
				for (i = 1; i < entryvec->n; i++)
					v.spl_left[i - 1] = i;
				gistToRealOffset(v.spl_left, v.spl_nleft, realoffset);
				v.spl_lattr[0] = v.spl_ldatum = (Datum) 0;
				v.spl_rattr[0] = v.spl_rdatum = (Datum) 0;
				v.spl_lisnull[0] = true;
				v.spl_risnull[0] = true;
				gistunionsubkey(r, giststate, itup, &v, true);
				v.spl_leftvalid = true;
				v.spl_rightvalid = false;
			}
		}
	}
	else
	{
		/* there is no invalid tuples, so usial processing */
		gistUserPicksplit(r, entryvec, &v, itup, *len, giststate);
		v.spl_leftvalid = v.spl_rightvalid = true;
	}


	/* form left and right vector */
	lvectup = (IndexTuple *) palloc(sizeof(IndexTuple) * (*len + 1));
	rvectup = (IndexTuple *) palloc(sizeof(IndexTuple) * (*len + 1));

	for (i = 0; i < v.spl_nleft; i++)
		lvectup[i] = itup[v.spl_left[i] - 1];

	for (i = 0; i < v.spl_nright; i++)
		rvectup[i] = itup[v.spl_right[i] - 1];

	/* place invalid tuples on right page if itsn't done yet */
	for (fakeoffset = entryvec->n; fakeoffset < *len + 1 && lencleaneditup; fakeoffset++)
	{
		rvectup[v.spl_nright++] = itup[realoffset[fakeoffset] - 1];
	}

	/* write on disk (may need another split) */
	if (gistnospace(right, rvectup, v.spl_nright))
	{
		nlen = v.spl_nright;
		newtup = gistSplit(r, rightbuf, rvectup, &nlen, dist, giststate);
		/* ReleaseBuffer(rightbuf); */
	}
	else
	{
		char	   *ptr;

		gistfillbuffer(r, right, rvectup, v.spl_nright, FirstOffsetNumber);
		/* XLOG stuff */
		ROTATEDIST(*dist);
		(*dist)->block.blkno = BufferGetBlockNumber(rightbuf);
		(*dist)->block.num = v.spl_nright;
		(*dist)->list = (IndexTupleData *) palloc(BLCKSZ);
		ptr = (char *) ((*dist)->list);
		for (i = 0; i < v.spl_nright; i++)
		{
			memcpy(ptr, rvectup[i], IndexTupleSize(rvectup[i]));
			ptr += IndexTupleSize(rvectup[i]);
		}
		(*dist)->lenlist = ptr - ((char *) ((*dist)->list));
		(*dist)->buffer = rightbuf;

		nlen = 1;
		newtup = (IndexTuple *) palloc(sizeof(IndexTuple) * 1);
		newtup[0] = (v.spl_rightvalid) ? gistFormTuple(giststate, r, v.spl_rattr, v.spl_rattrsize, v.spl_risnull)
			: gist_form_invalid_tuple(rbknum);
		ItemPointerSetBlockNumber(&(newtup[0]->t_tid), rbknum);
	}

	if (gistnospace(left, lvectup, v.spl_nleft))
	{
		int			llen = v.spl_nleft;
		IndexTuple *lntup;

		lntup = gistSplit(r, leftbuf, lvectup, &llen, dist, giststate);
		/* ReleaseBuffer(leftbuf); */

		newtup = gistjoinvector(newtup, &nlen, lntup, llen);
	}
	else
	{
		char	   *ptr;

		gistfillbuffer(r, left, lvectup, v.spl_nleft, FirstOffsetNumber);
		/* XLOG stuff */
		ROTATEDIST(*dist);
		(*dist)->block.blkno = BufferGetBlockNumber(leftbuf);
		(*dist)->block.num = v.spl_nleft;
		(*dist)->list = (IndexTupleData *) palloc(BLCKSZ);
		ptr = (char *) ((*dist)->list);
		for (i = 0; i < v.spl_nleft; i++)
		{
			memcpy(ptr, lvectup[i], IndexTupleSize(lvectup[i]));
			ptr += IndexTupleSize(lvectup[i]);
		}
		(*dist)->lenlist = ptr - ((char *) ((*dist)->list));
		(*dist)->buffer = leftbuf;

		if (BufferGetBlockNumber(buffer) != GIST_ROOT_BLKNO)
			PageRestoreTempPage(left, p);

		nlen += 1;
		newtup = (IndexTuple *) repalloc(newtup, sizeof(IndexTuple) * nlen);
		newtup[nlen - 1] = (v.spl_leftvalid) ? gistFormTuple(giststate, r, v.spl_lattr, v.spl_lattrsize, v.spl_lisnull)
			: gist_form_invalid_tuple(lbknum);
		ItemPointerSetBlockNumber(&(newtup[nlen - 1]->t_tid), lbknum);
	}

	GistClearTuplesDeleted(p);

	*len = nlen;
	return newtup;
}

void
gistnewroot(Relation r, Buffer buffer, IndexTuple *itup, int len, ItemPointer key)
{
	Page		page;

	Assert(BufferGetBlockNumber(buffer) == GIST_ROOT_BLKNO);
	page = BufferGetPage(buffer);
	GISTInitBuffer(buffer, 0);

	gistfillbuffer(r, page, itup, len, FirstOffsetNumber);
	if (!r->rd_istemp)
	{
		XLogRecPtr	recptr;
		XLogRecData *rdata;

		rdata = formUpdateRdata(r->rd_node, GIST_ROOT_BLKNO,
								NULL, 0, false, itup, len, key);

		START_CRIT_SECTION();

		recptr = XLogInsert(RM_GIST_ID, XLOG_GIST_NEW_ROOT, rdata);
		PageSetLSN(page, recptr);
		PageSetTLI(page, ThisTimeLineID);

		END_CRIT_SECTION();
	}
	else
		PageSetLSN(page, XLogRecPtrForTemp);
}

void
initGISTstate(GISTSTATE *giststate, Relation index)
{
	int			i;

	if (index->rd_att->natts > INDEX_MAX_KEYS)
		elog(ERROR, "numberOfAttributes %d > %d",
			 index->rd_att->natts, INDEX_MAX_KEYS);

	giststate->tupdesc = index->rd_att;

	for (i = 0; i < index->rd_att->natts; i++)
	{
		fmgr_info_copy(&(giststate->consistentFn[i]),
					   index_getprocinfo(index, i + 1, GIST_CONSISTENT_PROC),
					   CurrentMemoryContext);
		fmgr_info_copy(&(giststate->unionFn[i]),
					   index_getprocinfo(index, i + 1, GIST_UNION_PROC),
					   CurrentMemoryContext);
		fmgr_info_copy(&(giststate->compressFn[i]),
					   index_getprocinfo(index, i + 1, GIST_COMPRESS_PROC),
					   CurrentMemoryContext);
		fmgr_info_copy(&(giststate->decompressFn[i]),
					   index_getprocinfo(index, i + 1, GIST_DECOMPRESS_PROC),
					   CurrentMemoryContext);
		fmgr_info_copy(&(giststate->penaltyFn[i]),
					   index_getprocinfo(index, i + 1, GIST_PENALTY_PROC),
					   CurrentMemoryContext);
		fmgr_info_copy(&(giststate->picksplitFn[i]),
					   index_getprocinfo(index, i + 1, GIST_PICKSPLIT_PROC),
					   CurrentMemoryContext);
		fmgr_info_copy(&(giststate->equalFn[i]),
					   index_getprocinfo(index, i + 1, GIST_EQUAL_PROC),
					   CurrentMemoryContext);
	}
}

void
freeGISTstate(GISTSTATE *giststate)
{
	/* no work */
}