tuplesort.c

来自「PostgreSQL7.4.6 for Linux」· C语言 代码 · 共 2,177 行 · 第 1/5 页

	return state;
}

Tuplesortstate *
tuplesort_begin_heap(TupleDesc tupDesc,
					 int nkeys,
					 Oid *sortOperators, AttrNumber *attNums,
					 bool randomAccess)
{
	Tuplesortstate *state = tuplesort_begin_common(randomAccess);
	int			i;

	AssertArg(nkeys > 0);

	state->comparetup = comparetup_heap;
	state->copytup = copytup_heap;
	state->writetup = writetup_heap;
	state->readtup = readtup_heap;

	state->tupDesc = tupDesc;
	state->nKeys = nkeys;
	state->scanKeys = (ScanKey) palloc0(nkeys * sizeof(ScanKeyData));
	state->sortFnKinds = (SortFunctionKind *)
		palloc0(nkeys * sizeof(SortFunctionKind));

	for (i = 0; i < nkeys; i++)
	{
		RegProcedure sortFunction;

		AssertArg(sortOperators[i] != 0);
		AssertArg(attNums[i] != 0);

		/* select a function that implements the sort operator */
		SelectSortFunction(sortOperators[i], &sortFunction,
						   &state->sortFnKinds[i]);

		ScanKeyEntryInitialize(&state->scanKeys[i],
							   0x0,
							   attNums[i],
							   sortFunction,
							   (Datum) 0);
	}

	return state;
}

Tuplesortstate *
tuplesort_begin_index(Relation indexRel,
					  bool enforceUnique,
					  bool randomAccess)
{
	Tuplesortstate *state = tuplesort_begin_common(randomAccess);

	state->comparetup = comparetup_index;
	state->copytup = copytup_index;
	state->writetup = writetup_index;
	state->readtup = readtup_index;

	state->indexRel = indexRel;
	/* see comments below about btree dependence of this code... */
	state->indexScanKey = _bt_mkscankey_nodata(indexRel);
	state->enforceUnique = enforceUnique;

	return state;
}

Tuplesortstate *
tuplesort_begin_datum(Oid datumType,
					  Oid sortOperator,
					  bool randomAccess)
{
	Tuplesortstate *state = tuplesort_begin_common(randomAccess);
	RegProcedure sortFunction;
	int16		typlen;
	bool		typbyval;

	state->comparetup = comparetup_datum;
	state->copytup = copytup_datum;
	state->writetup = writetup_datum;
	state->readtup = readtup_datum;

	state->datumType = datumType;
	state->sortOperator = sortOperator;

	/* select a function that implements the sort operator */
	SelectSortFunction(sortOperator, &sortFunction, &state->sortFnKind);
	/* and look up the function */
	fmgr_info(sortFunction, &state->sortOpFn);

	/* lookup necessary attributes of the datum type */
	get_typlenbyval(datumType, &typlen, &typbyval);
	state->datumTypeLen = typlen;
	state->datumTypeByVal = typbyval;

	return state;
}

/*
 * tuplesort_end
 *
 *	Release resources and clean up.
 */
void
tuplesort_end(Tuplesortstate *state)
{
	int			i;

	if (state->tapeset)
		LogicalTapeSetClose(state->tapeset);
	if (state->memtuples)
	{
		for (i = 0; i < state->memtupcount; i++)
			pfree(state->memtuples[i]);
		pfree(state->memtuples);
	}
	if (state->memtupindex)
		pfree(state->memtupindex);

	/*
	 * this stuff might better belong in a variant-specific shutdown
	 * routine
	 */
	if (state->scanKeys)
		pfree(state->scanKeys);
	if (state->sortFnKinds)
		pfree(state->sortFnKinds);

	pfree(state);
}

/*
 * Accept one tuple while collecting input data for sort.
 *
 * Note that the input tuple is always copied; the caller need not save it.
 */
void
tuplesort_puttuple(Tuplesortstate *state, void *tuple)
{
	/*
	 * Copy the given tuple into memory we control, and decrease availMem.
	 * Then call the code shared with the Datum case.
	 */
	tuple = COPYTUP(state, tuple);

	puttuple_common(state, tuple);
}

/*
 * Accept one Datum while collecting input data for sort.
 *
 * If the Datum is pass-by-ref type, the value will be copied.
 */
void
tuplesort_putdatum(Tuplesortstate *state, Datum val, bool isNull)
{
	DatumTuple *tuple;

	/*
	 * Build pseudo-tuple carrying the datum, and decrease availMem.
	 */
	if (isNull || state->datumTypeByVal)
	{
		tuple = (DatumTuple *) palloc(sizeof(DatumTuple));
		tuple->val = val;
		tuple->isNull = isNull;
	}
	else
	{
		Size		datalen;
		Size		tuplelen;
		char	   *newVal;

		datalen = datumGetSize(val, false, state->datumTypeLen);
		tuplelen = datalen + MAXALIGN(sizeof(DatumTuple));
		tuple = (DatumTuple *) palloc(tuplelen);
		newVal = ((char *) tuple) + MAXALIGN(sizeof(DatumTuple));
		memcpy(newVal, DatumGetPointer(val), datalen);
		tuple->val = PointerGetDatum(newVal);
		tuple->isNull = false;
	}

	USEMEM(state, GetMemoryChunkSpace(tuple));

	puttuple_common(state, (void *) tuple);
}

/*
 * Shared code for tuple and datum cases.
 */
static void
puttuple_common(Tuplesortstate *state, void *tuple)
{
	switch (state->status)
	{
		case TSS_INITIAL:

			/*
			 * Save the copied tuple into the unsorted array.
			 */
			if (state->memtupcount >= state->memtupsize)
			{
				/* Grow the unsorted array as needed. */
				FREEMEM(state, GetMemoryChunkSpace(state->memtuples));
				state->memtupsize *= 2;
				state->memtuples = (void **)
					repalloc(state->memtuples,
							 state->memtupsize * sizeof(void *));
				USEMEM(state, GetMemoryChunkSpace(state->memtuples));
			}
			state->memtuples[state->memtupcount++] = tuple;

			/*
			 * Done if we still fit in available memory.
			 */
			if (!LACKMEM(state))
				return;

			/*
			 * Nope; time to switch to tape-based operation.
			 */
			inittapes(state);

			/*
			 * Dump tuples until we are back under the limit.
			 */
			dumptuples(state, false);
			break;
		case TSS_BUILDRUNS:

			/*
			 * Insert the copied tuple into the heap, with run number
			 * currentRun if it can go into the current run, else run
			 * number currentRun+1.  The tuple can go into the current run
			 * if it is >= the first not-yet-output tuple.	(Actually, it
			 * could go into the current run if it is >= the most recently
			 * output tuple ... but that would require keeping around the
			 * tuple we last output, and it's simplest to let writetup
			 * free each tuple as soon as it's written.)
			 *
			 * Note there will always be at least one tuple in the heap at
			 * this point; see dumptuples.
			 */
			Assert(state->memtupcount > 0);
			if (COMPARETUP(state, tuple, state->memtuples[0]) >= 0)
				tuplesort_heap_insert(state, tuple, state->currentRun, true);
			else
				tuplesort_heap_insert(state, tuple, state->currentRun + 1, true);

			/*
			 * If we are over the memory limit, dump tuples till we're
			 * under.
			 */
			dumptuples(state, false);
			break;
		default:
			elog(ERROR, "invalid tuplesort state");
			break;
	}
}

/*
 * All tuples have been provided; finish the sort.
 */
void
tuplesort_performsort(Tuplesortstate *state)
{
	switch (state->status)
	{
		case TSS_INITIAL:

			/*
			 * We were able to accumulate all the tuples within the
			 * allowed amount of memory.  Just qsort 'em and we're done.
			 */
			if (state->memtupcount > 1)
			{
				qsort_tuplesortstate = state;
				qsort((void *) state->memtuples, state->memtupcount,
					  sizeof(void *), qsort_comparetup);
			}
			state->current = 0;
			state->eof_reached = false;
			state->markpos_offset = 0;
			state->markpos_eof = false;
			state->status = TSS_SORTEDINMEM;
			break;
		case TSS_BUILDRUNS:

			/*
			 * Finish tape-based sort.	First, flush all tuples remaining
			 * in memory out to tape; then merge until we have a single
			 * remaining run (or, if !randomAccess, one run per tape).
			 * Note that mergeruns sets the correct state->status.
			 */
			dumptuples(state, true);
			mergeruns(state);
			state->eof_reached = false;
			state->markpos_block = 0L;
			state->markpos_offset = 0;
			state->markpos_eof = false;
			break;
		default:
			elog(ERROR, "invalid tuplesort state");
			break;
	}
}

/*
 * Fetch the next tuple in either forward or back direction.
 * Returns NULL if no more tuples.	If should_free is set, the
 * caller must pfree the returned tuple when done with it.
 */
void *
tuplesort_gettuple(Tuplesortstate *state, bool forward,
				   bool *should_free)
{
	unsigned int tuplen;
	void	   *tup;

	switch (state->status)
	{
		case TSS_SORTEDINMEM:
			Assert(forward || state->randomAccess);
			*should_free = false;
			if (forward)
			{
				if (state->current < state->memtupcount)
					return state->memtuples[state->current++];
				state->eof_reached = true;
				return NULL;
			}
			else
			{
				if (state->current <= 0)
					return NULL;

				/*
				 * if all tuples are fetched already then we return last
				 * tuple, else - tuple before last returned.
				 */
				if (state->eof_reached)
					state->eof_reached = false;
				else
				{
					state->current--;	/* last returned tuple */
					if (state->current <= 0)
						return NULL;
				}
				return state->memtuples[state->current - 1];
			}
			break;

		case TSS_SORTEDONTAPE:
			Assert(forward || state->randomAccess);
			*should_free = true;
			if (forward)
			{
				if (state->eof_reached)
					return NULL;
				if ((tuplen = getlen(state, state->result_tape, true)) != 0)
				{
					tup = READTUP(state, state->result_tape, tuplen);
					return tup;
				}
				else
				{
					state->eof_reached = true;
					return NULL;
				}
			}

			/*
			 * Backward.
			 *
			 * if all tuples are fetched already then we return last tuple,
			 * else - tuple before last returned.
			 */
			if (state->eof_reached)
			{
				/*
				 * Seek position is pointing just past the zero tuplen at
				 * the end of file; back up to fetch last tuple's ending
				 * length word.  If seek fails we must have a completely
				 * empty file.
				 */
				if (!LogicalTapeBackspace(state->tapeset,
										  state->result_tape,
										  2 * sizeof(unsigned int)))
					return NULL;
				state->eof_reached = false;
			}
			else
			{
				/*
				 * Back up and fetch previously-returned tuple's ending
				 * length word.  If seek fails, assume we are at start of
				 * file.
				 */
				if (!LogicalTapeBackspace(state->tapeset,
										  state->result_tape,
										  sizeof(unsigned int)))
					return NULL;
				tuplen = getlen(state, state->result_tape, false);

				/*
				 * Back up to get ending length word of tuple before it.
				 */
				if (!LogicalTapeBackspace(state->tapeset,
										  state->result_tape,
									  tuplen + 2 * sizeof(unsigned int)))
				{
					/*
					 * If that fails, presumably the prev tuple is the
					 * first in the file.  Back up so that it becomes next
					 * to read in forward direction (not obviously right,
					 * but that is what in-memory case does).
					 */
					if (!LogicalTapeBackspace(state->tapeset,
											  state->result_tape,
										  tuplen + sizeof(unsigned int)))
						elog(ERROR, "bogus tuple length in backward scan");
					return NULL;
				}
			}

			tuplen = getlen(state, state->result_tape, false);

			/*
			 * Now we have the length of the prior tuple, back up and read
			 * it. Note: READTUP expects we are positioned after the
			 * initial length word of the tuple, so back up to that point.
			 */
			if (!LogicalTapeBackspace(state->tapeset,
									  state->result_tape,
									  tuplen))