tuplesort.c

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

						memtuples[j], memtupindex[j]) <= 0)
			break;
		memtuples[i] = memtuples[j];
		memtupindex[i] = memtupindex[j];
		i = j;
	}
	memtuples[i] = tuple;
	memtupindex[i] = tupindex;
}


/*
 * Tape interface routines
 */

static unsigned int
getlen(Tuplesortstate *state, int tapenum, bool eofOK)
{
	unsigned int len;

	if (LogicalTapeRead(state->tapeset, tapenum, (void *) &len,
						sizeof(len)) != sizeof(len))
		elog(ERROR, "unexpected end of tape");
	if (len == 0 && !eofOK)
		elog(ERROR, "unexpected end of data");
	return len;
}

static void
markrunend(Tuplesortstate *state, int tapenum)
{
	unsigned int len = 0;

	LogicalTapeWrite(state->tapeset, tapenum, (void *) &len, sizeof(len));
}


/*
 * qsort interface
 */

static int
qsort_comparetup(const void *a, const void *b)
{
	/* The passed pointers are pointers to void * ... */

	return COMPARETUP(qsort_tuplesortstate, *(void **) a, *(void **) b);
}


/*
 * This routine selects an appropriate sorting function to implement
 * a sort operator as efficiently as possible.	The straightforward
 * method is to use the operator's implementation proc --- ie, "<"
 * comparison.	However, that way often requires two calls of the function
 * per comparison.	If we can find a btree three-way comparator function
 * associated with the operator, we can use it to do the comparisons
 * more efficiently.  We also support the possibility that the operator
 * is ">" (descending sort), in which case we have to reverse the output
 * of the btree comparator.
 *
 * Possibly this should live somewhere else (backend/catalog/, maybe?).
 */
void
SelectSortFunction(Oid sortOperator,
				   RegProcedure *sortFunction,
				   SortFunctionKind *kind)
{
	CatCList   *catlist;
	int			i;
	HeapTuple	tuple;
	Form_pg_operator optup;
	Oid			opclass = InvalidOid;

	/*
	 * Search pg_amop to see if the target operator is registered as the "<"
	 * or ">" operator of any btree opclass.  It's possible that it might be
	 * registered both ways (eg, if someone were to build a "reverse sort"
	 * opclass for some reason); prefer the "<" case if so. If the operator is
	 * registered the same way in multiple opclasses, assume we can use the
	 * associated comparator function from any one.
	 */
	catlist = SearchSysCacheList(AMOPOPID, 1,
								 ObjectIdGetDatum(sortOperator),
								 0, 0, 0);

	for (i = 0; i < catlist->n_members; i++)
	{
		Form_pg_amop aform;

		tuple = &catlist->members[i]->tuple;
		aform = (Form_pg_amop) GETSTRUCT(tuple);

		if (!opclass_is_btree(aform->amopclaid))
			continue;
		/* must be of default subtype, too */
		if (aform->amopsubtype != InvalidOid)
			continue;

		if (aform->amopstrategy == BTLessStrategyNumber)
		{
			opclass = aform->amopclaid;
			*kind = SORTFUNC_CMP;
			break;				/* done looking */
		}
		else if (aform->amopstrategy == BTGreaterStrategyNumber)
		{
			opclass = aform->amopclaid;
			*kind = SORTFUNC_REVCMP;
			/* keep scanning in hopes of finding a BTLess entry */
		}
	}

	ReleaseSysCacheList(catlist);

	if (OidIsValid(opclass))
	{
		/* Found a suitable opclass, get its default comparator function */
		*sortFunction = get_opclass_proc(opclass, InvalidOid, BTORDER_PROC);
		Assert(RegProcedureIsValid(*sortFunction));
		return;
	}

	/*
	 * Can't find a comparator, so use the operator as-is.  Decide whether it
	 * is forward or reverse sort by looking at its name (grotty, but this
	 * only matters for deciding which end NULLs should get sorted to).  XXX
	 * possibly better idea: see whether its selectivity function is
	 * scalargtcmp?
	 */
	tuple = SearchSysCache(OPEROID,
						   ObjectIdGetDatum(sortOperator),
						   0, 0, 0);
	if (!HeapTupleIsValid(tuple))
		elog(ERROR, "cache lookup failed for operator %u", sortOperator);
	optup = (Form_pg_operator) GETSTRUCT(tuple);
	if (strcmp(NameStr(optup->oprname), ">") == 0)
		*kind = SORTFUNC_REVLT;
	else
		*kind = SORTFUNC_LT;
	*sortFunction = optup->oprcode;
	ReleaseSysCache(tuple);

	Assert(RegProcedureIsValid(*sortFunction));
}

/*
 * Inline-able copy of FunctionCall2() to save some cycles in sorting.
 */
static inline Datum
myFunctionCall2(FmgrInfo *flinfo, Datum arg1, Datum arg2)
{
	FunctionCallInfoData fcinfo;
	Datum		result;

	InitFunctionCallInfoData(fcinfo, flinfo, 2, NULL, NULL);

	fcinfo.arg[0] = arg1;
	fcinfo.arg[1] = arg2;
	fcinfo.argnull[0] = false;
	fcinfo.argnull[1] = false;

	result = FunctionCallInvoke(&fcinfo);

	/* Check for null result, since caller is clearly not expecting one */
	if (fcinfo.isnull)
		elog(ERROR, "function %u returned NULL", fcinfo.flinfo->fn_oid);

	return result;
}

/*
 * Apply a sort function (by now converted to fmgr lookup form)
 * and return a 3-way comparison result.  This takes care of handling
 * NULLs and sort ordering direction properly.
 */
static inline int32
inlineApplySortFunction(FmgrInfo *sortFunction, SortFunctionKind kind,
						Datum datum1, bool isNull1,
						Datum datum2, bool isNull2)
{
	switch (kind)
	{
		case SORTFUNC_LT:
			if (isNull1)
			{
				if (isNull2)
					return 0;
				return 1;		/* NULL sorts after non-NULL */
			}
			if (isNull2)
				return -1;
			if (DatumGetBool(myFunctionCall2(sortFunction, datum1, datum2)))
				return -1;		/* a < b */
			if (DatumGetBool(myFunctionCall2(sortFunction, datum2, datum1)))
				return 1;		/* a > b */
			return 0;

		case SORTFUNC_REVLT:
			/* We reverse the ordering of NULLs, but not the operator */
			if (isNull1)
			{
				if (isNull2)
					return 0;
				return -1;		/* NULL sorts before non-NULL */
			}
			if (isNull2)
				return 1;
			if (DatumGetBool(myFunctionCall2(sortFunction, datum1, datum2)))
				return -1;		/* a < b */
			if (DatumGetBool(myFunctionCall2(sortFunction, datum2, datum1)))
				return 1;		/* a > b */
			return 0;

		case SORTFUNC_CMP:
			if (isNull1)
			{
				if (isNull2)
					return 0;
				return 1;		/* NULL sorts after non-NULL */
			}
			if (isNull2)
				return -1;
			return DatumGetInt32(myFunctionCall2(sortFunction,
												 datum1, datum2));

		case SORTFUNC_REVCMP:
			if (isNull1)
			{
				if (isNull2)
					return 0;
				return -1;		/* NULL sorts before non-NULL */
			}
			if (isNull2)
				return 1;
			return -DatumGetInt32(myFunctionCall2(sortFunction,
												  datum1, datum2));

		default:
			elog(ERROR, "unrecognized SortFunctionKind: %d", (int) kind);
			return 0;			/* can't get here, but keep compiler quiet */
	}
}

/*
 * Non-inline ApplySortFunction() --- this is needed only to conform to
 * C99's brain-dead notions about how to implement inline functions...
 */
int32
ApplySortFunction(FmgrInfo *sortFunction, SortFunctionKind kind,
				  Datum datum1, bool isNull1,
				  Datum datum2, bool isNull2)
{
	return inlineApplySortFunction(sortFunction, kind,
								   datum1, isNull1,
								   datum2, isNull2);
}


/*
 * Routines specialized for HeapTuple case
 */

static int
comparetup_heap(Tuplesortstate *state, const void *a, const void *b)
{
	HeapTuple	ltup = (HeapTuple) a;
	HeapTuple	rtup = (HeapTuple) b;
	TupleDesc	tupDesc = state->tupDesc;
	int			nkey;

	for (nkey = 0; nkey < state->nKeys; nkey++)
	{
		ScanKey		scanKey = state->scanKeys + nkey;
		AttrNumber	attno = scanKey->sk_attno;
		Datum		datum1,
					datum2;
		bool		isnull1,
					isnull2;
		int32		compare;

		datum1 = heap_getattr(ltup, attno, tupDesc, &isnull1);
		datum2 = heap_getattr(rtup, attno, tupDesc, &isnull2);

		compare = inlineApplySortFunction(&scanKey->sk_func,
										  state->sortFnKinds[nkey],
										  datum1, isnull1,
										  datum2, isnull2);
		if (compare != 0)
			return compare;
	}

	return 0;
}

static void *
copytup_heap(Tuplesortstate *state, void *tup)
{
	HeapTuple	tuple = (HeapTuple) tup;

	tuple = heap_copytuple(tuple);
	USEMEM(state, GetMemoryChunkSpace(tuple));
	return (void *) tuple;
}

/*
 * We don't bother to write the HeapTupleData part of the tuple.
 */

static void
writetup_heap(Tuplesortstate *state, int tapenum, void *tup)
{
	HeapTuple	tuple = (HeapTuple) tup;
	unsigned int tuplen;

	tuplen = tuple->t_len + sizeof(tuplen);
	LogicalTapeWrite(state->tapeset, tapenum,
					 (void *) &tuplen, sizeof(tuplen));
	LogicalTapeWrite(state->tapeset, tapenum,
					 (void *) tuple->t_data, tuple->t_len);
	if (state->randomAccess)	/* need trailing length word? */
		LogicalTapeWrite(state->tapeset, tapenum,
						 (void *) &tuplen, sizeof(tuplen));

	FREEMEM(state, GetMemoryChunkSpace(tuple));
	heap_freetuple(tuple);
}

static void *
readtup_heap(Tuplesortstate *state, int tapenum, unsigned int len)
{
	unsigned int tuplen = len - sizeof(unsigned int) + HEAPTUPLESIZE;
	HeapTuple	tuple = (HeapTuple) palloc(tuplen);

	USEMEM(state, GetMemoryChunkSpace(tuple));
	/* reconstruct the HeapTupleData portion */
	tuple->t_len = len - sizeof(unsigned int);
	ItemPointerSetInvalid(&(tuple->t_self));
	tuple->t_datamcxt = CurrentMemoryContext;
	tuple->t_data = (HeapTupleHeader) (((char *) tuple) + HEAPTUPLESIZE);
	/* read in the tuple proper */
	if (LogicalTapeRead(state->tapeset, tapenum, (void *) tuple->t_data,
						tuple->t_len) != tuple->t_len)
		elog(ERROR, "unexpected end of data");
	if (state->randomAccess)	/* need trailing length word? */
		if (LogicalTapeRead(state->tapeset, tapenum, (void *) &tuplen,
							sizeof(tuplen)) != sizeof(tuplen))
			elog(ERROR, "unexpected end of data");
	return (void *) tuple;
}


/*
 * Routines specialized for IndexTuple case
 *
 * NOTE: actually, these are specialized for the btree case; it's not
 * clear whether you could use them for a non-btree index.	Possibly
 * you'd need to make another set of routines if you needed to sort
 * according to another kind of index.
 */

static int
comparetup_index(Tuplesortstate *state, const void *a, const void *b)
{
	/*
	 * This is almost the same as _bt_tuplecompare(), but we need to keep
	 * track of whether any null fields are present.  Also see the special
	 * treatment for equal keys at the end.
	 */
	IndexTuple	tuple1 = (IndexTuple) a;
	IndexTuple	tuple2 = (IndexTuple) b;
	Relation	rel = state->indexRel;
	int			keysz = RelationGetNumberOfAttributes(rel);
	ScanKey		scankey = state->indexScanKey;
	TupleDesc	tupDes;
	int			i;
	bool		equal_hasnull = false;

	tupDes = RelationGetDescr(rel);

	for (i = 1; i <= keysz; i++)
	{
		ScanKey		entry = &scankey[i - 1];
		Datum		datum1,
					datum2;
		bool		isnull1,
					isnull2;
		int32		compare;

		datum1 = index_getattr(tuple1, i, tupDes, &isnull1);
		datum2 = index_getattr(tuple2, i, tupDes, &isnull2);

		/* see comments about NULLs handling in btbuild */

		/* the comparison function is always of CMP type */
		compare = inlineApplySortFunction(&entry->sk_func, SORTFUNC_CMP,
										  datum1, isnull1,
										  datum2, isnull2);

		if (compare != 0)
			return (int) compare;		/* done when we find unequal
										 * attributes */

		/* they are equal, so we only need to examine one null flag */
		if (isnull1)
			equal_hasnull = true;
	}

	/*
	 * If btree has asked us to enforce uniqueness, complain if two equal
	 * tuples are detected (unless there was at least one NULL field).
	 *
	 * It is sufficient to make the test here, because if two tuples are equal
	 * they *must* get compared at some stage of the sort --- otherwise the
	 * sort algorithm wouldn't have checked whether one must appear before the
	 * other.
	 *
	 * Some rather brain-dead implementations of qsort will sometimes call the
	 * comparison routine to compare a value to itself.  (At this writing only
	 * QNX 4 is known to do such silly things.)  Don't raise a bogus error in
	 * that case.
	 */
	if (state->enforceUnique && !equal_hasnull && tuple1 != tuple2)
		ereport(ERROR,
				(errcode(ERRCODE_UNIQUE_VIOLATION),
				 errmsg("could not create unique index"),
				 errdetail("Table contains duplicated values.")));

	/*
	 * If key values are equal, we sort on ItemPointer.  This does not affect
	 * validity of the finished index, but it offers cheap insurance agains